impCentral API Example Code: Device Logging via Stream

logs.js

// CONSTANTS
// Replace the 'USERNAME' and 'PASSWORD' values with your own, and enter
// Replace the 'DEVICE_ID' value with the ID of a device accessible via impCentral (not the IDE)
// Replace the 'ACCESS_TOKEN' value with one retrieved using 'products.js' at
// https://gist.github.com/ElectricImpSampleCode/f0fb7f8bd04be009beeb5ee86dbd406a 
const USERNAME = '...';
const PASSWORD = '...';
const DEVICE_ID = '...';
const ACCESS_TOKEN = '...';
const API_URL = 'api.electricimp.com';
const API_VER = '/v5/';

var https = require('https');
var loggingStarted = false;

// FUNCTIONS
function setOptions(verb, path, contentType, token) {
  // Returns an HTTPS request options object primed for API usage
  return {
    "hostname": API_URL,
    "path": API_VER + path,
    "method": verb,
    "headers": {
      'Content-Type': contentType,
      'Authorization': 'Bearer ' + token
    }
  };
}

function showError(resp) {
  console.log(`STATUS: ${resp.statusCode}`);
  console.log(`HEADERS: ${JSON.stringify(resp.headers)}`);
}

function initStream() {
  // Set up a request to initialize the account's log strem
  let req = https.request(setOptions('POST', 'logstream', 'application/vnd.api+json', ACCESS_TOKEN), (resp) => {
    let body = '';
    resp.setEncoding('utf8');
    resp.on('data', (chunk) => { body += chunk; });
    resp.on('end', () => {
      if (body.length > 0) {
        try {
          let data = JSON.parse(body);
          data = data.data;
          if (resp.statusCode === 200) {
            // Log stream initialized - report and then open it
            console.log(`Stream initialized at ${data.attributes.url}`);
            beginStream(data.attributes.url);
          } else {
            // API Error
            console.log(`API ERROR: ${body}`);
          }
        } catch (err) {
          // JSON Error
          console.log(err);
        }
      } else {
        // HTTP Error
        showError(resp);
      }
    });
  });

  req.on('error', (err) => { console.error(`REQUEST ERROR: ${err.message}`); });
  req.end();
}

function beginStream(url) {
  // Set up the stream. We need to parse the URL returned by the impCentral API
  // into the separate components required by Node's http.request()
  let urlarray = url.split('//');

  // Remove the 'https:'
  urlarray = urlarray[1].split('/');

  let path = '';
  let count = 0;
  for (part of urlarray) {
    // Ignore the first two URL - the host name and API version number
    if (count > 1) {
      path += part + '/';
    }

    count += 1;
  }

  let messageBuffer = '';
  let logreq = https.request(setOptions('GET', path, 'text/event-stream', ACCESS_TOKEN), (resp) => {
    // We have the stream running - do we need to register a device's ID?
    if (loggingStarted === false) {
          // Add device to stream
          addDevice(path, DEVICE_ID);
          loggingStarted = true;
    }

    resp.setEncoding('utf8');

    resp.on('data', (chunk) => {
      // For logging, all SSE messages are returned as chunks, so we process the incoming data
      // here not during the 'end' callback
      messageBuffer += chunk;
      let i = -1;
      do {
        i = messageBuffer.indexOf('\n\n');
        if (i !== -1) {
          // We have at least one message in the received text, so get it...
          let message = messageBuffer.substring(0, i);
          
          // ...and remove it from the messzge buffer
          messageBuffer = messageBuffer.substring(i + 2);
          processMessage(message);
        }
      } while (i !== -1 && messageBuffer.length > 0);
    });

    resp.on('end', () => {
      console.log("END");
    });
  });

  logreq.on('error', (err) => { console.error(`REQUEST ERROR: ${err.message}`); });
  logreq.end();
}

function processMessage(message) {
  // Only deal with SSE messages - first, segment message by end-of-field markers
  let fields = message.split('\n');
  
  // Remove initial 'event: ' and 'data: ' from the fields
  let event = fields[0].substring(fields[0].indexOf(': ') + 2);
  let data = fields.length > 1 ? fields[1].substring(fields[1].indexOf(': ') + 2) : '';

  if (event === 'message') {
    // Display the message itself in the style of the impCentral log
    if (data.length > 0) {
      let da = data.split(' ');
      data = '';
      
      for (let i = 4 ; i < da.length ; i++) {
        data = data + da[i] + ' ';
      }

      console.log('[ITEM] ' + da[1] + ' [' + da[3] + '] ' + data);
    }
  } else if (event === 'state_change') {
    // Display a status message
    if (data.length > 0) {
      console.log('[INFO] ' + data);
    }
  }
}

function addDevice(path, id) {
  // Add the device to the stream with a PUT request
  let req = https.request(setOptions('PUT', path + id, 'application/vnd.api+json', ACCESS_TOKEN), (resp) => {
    let body ='';
    resp.setEncoding('utf8');
    resp.on('data', (chunk) => { body += chunk });
    resp.on('end', () => {
      if (resp.statusCode !== 204) {
        // Status code 204 indicates device added
        console.error(`Could not add device ${id} to the stream (${resp.statusCode})`);
      }
    });
  });

  // Write the payload to the request
  req.write(JSON.stringify({
    'id': id,
    'type': 'device'
  }));

  req.on('error', (err) => { console.error(`REQUEST ERROR: ${err.message}`); });
  req.end();
}

// RUNTIME
// Initialize the log stream
initStream();