API Builder and MQTT for IoT - Part 2 (files)

apibmqttutils.js

var Arrow = require('arrow');

var client;

module.exports.mqttInit = function() {
	var mqtt = require('mqtt');
	var mqtturl = 'mqtt://<user>:<password>@<server>:<port>';
	client = mqtt.connect(mqtturl);
	
	client.on('connect', function () {
		console.log('app.js - client connected');
		client.subscribe('myhome/server/#');
	});
	
	client.on('message', function (topic, message) {
		console.log('app.js - message received');
		handleMessage(topic, message);
	});
};

var handleMessage = function(topic, message) {
	console.log("apibmqttutils.js - processMessage() - topic = "+topic+" message = "+message);
    
    switch(topic) {
    	case 'myhome/server':
    		processMessage(message);
    		break;
    	
    	case 'myhome/server/will':
    		processLWTMessage(message);
    		break;
    	
    	default: unhandledTopic(topic, message);
    }
};

var unhandledTopic = function(topic, message) {
	console.log('apibmqttutils.js - unhandledTopic() - Unhandled Topic: '+topic);
};

var getDevice = function(deviceId, callback) {
	console.log("apibmqttutils.js - getDevice() - deviceId = "+deviceId);
	
	var model = Arrow.getModel("device");
    model.query({deviceId: deviceId}, function(err, data){
  		if(err) {
  			console.log('apibmqttutils.js - getDevice() - error accessing iot device database, err = '+err);
  			if(callback) {callback(null)};
  		} else {
  			console.log('apibmqttutils.js - getDevice() - data.length = '+data.length);
  			console.log('apibmqttutils.js - getDevice() - data = '+JSON.stringify(data));
  			if(callback) {callback(data)};
  		}
  	});
};

var processLWTMessage = function(message){
	console.log("apibmqttutils.js - processLWTMessage() - message = "+message);
	
	getDevice(message.toString(), function(data){
		if(data) {
			if(data.length === 1) {
				data[0].isConnected = false;
				data[0].update();
			} else {
				console.log('apibmqttutils.js - processLWTMessage() - error with iot device, exactly 1 device not found');
			}
		} else {
			console.log('apibmqttutils.js - processLWTMessage() - error getting iot device');
		}
	});
};

var processMessage = function(message, callback) {
	console.log("apibmqttutils.js - processMessage() - message = "+message);
	
	message = JSON.parse(message);
	
	getDevice(message.deviceId, function(data){
		if(data) {
			console.log('apibmqttutils.js - processMessage() - data.length = '+data.length);
  			console.log('apibmqttutils.js - processMessage() - data = '+JSON.stringify(data));
			if(data.length > 1) {
  				console.log('apibmqttutils.js - processMessage() - Number of matching records greater than 1. Multiple devices with same deviceId present in device DB');
  			} else {
  				if(data.length === 1) {
  					console.log('apibmqttutils.js - processMessage() - Number of matching records equals to 1');
			        data[0].isConnected = true;
			        data[0].temp = message.temp.toString();
			        data[0].update();
			        console.log('apibmqttutils.js - processMessage() - Record updated!!!');
  				} else {
  					console.log('apibmqttutils.js - processMessage() - No matching records found, create new device record');
  				}
  			}
		} else {
			console.log('apibmqttutils.js - processMessage() - error getting iot device');
		}
	});
};

app.js

var Arrow = require('arrow'),
	server = new Arrow();
	
var apibmqttutils = require('./apibmqttutils').mqttInit();

// lifecycle examples
server.on('starting', function () {
	server.logger.debug('server is starting!');
});

server.on('started', function () {
	server.logger.debug('server started!');
});

// start the server
server.start();

device.js

var Arrow = require('arrow');
var Model = Arrow.createModel('device', {
    "fields": {
        "deviceId": {
            "type": "string",
            "description": "Device ID (mac address of IoT device)"
        },
        "isConnected": {
            "type": "boolean",
            "description": "Is the device connected? Set by the server"
        },
        "temp": {
            "type": "string",
            "description": "Temperature sensor reading"
        }
    },
    "connector": "appc.arrowdb",
    "actions": [
        "create",
        "read",
        "update",
        "delete",
        "deleteAll"
    ],
    "description": "IoT Device"
});
module.exports = Model;

index.js

var cloudMQTTUrl = 'mqtt://<user>:<password>@<server>:<port>';
var mqtt = require('mqtt');
var client;
var deviceId;

require('getmac').getMac(function(err, macAddress){
    if (err)  throw err;
    deviceId = macAddress;
    console.log(deviceId);
    client  = mqtt.connect(cloudMQTTUrl,
    {
      clientId: deviceId,
      will: {
        topic: 'myhome/server/will',
        payload: deviceId
      },
      keepalive: 60
    });

    client.on('connect', function () {
      console.log('client connected');
      message = JSON.stringify({
        deviceId:deviceId,
        temp : 70 + 5*Math.random() -2.5
      });
      client.publish('myhome/server', message);
    });

    setInterval(function(){
      console.log('sending ...');
      message = JSON.stringify({
        deviceId:deviceId,
        temp : 70 + 5*Math.random() -2.5
      });
      client.publish('myhome/server', message);
    }, 15000);

});

thanks for providing such a good sample.