yunkong2.mqtt/test/testClient.js

'use strict';
const expect  = require('chai').expect;
const setup   = require(__dirname + '/lib/setup');

let objects = null;
let states  = null;
let MqttServer;
let mqttClient = null;
let mqttServer = null;
let connected  = false;
let lastReceivedTopic;
let lastReceivedMessage;

const rules = {
    '/mqtt/0/test1': 'mqtt.0.test1',
    'mqtt/0/test2':  'mqtt.0.test2',
    'test3':         'mqtt.0.test3',
    'te s t4':       'mqtt.0.te_s_t4',
    'system/adapter/admin/upload':  'system.adapter.admin.upload',
    '/system/adapter/admin/upload': 'system.adapter.admin.upload'
};

function checkMqtt2Adapter(id, _expectedId, _it, _done) {
    _it.timeout(1000);
    const value = 'Roger' + Math.round(Math.random() * 100);
    const mqttid = id;
    if (!_expectedId) {
        id = id.replace(/\//g, '.').replace(/\s/g, '_');
        if (id[0] === '.') id = id.substring(1);
    } else {
        id = _expectedId;
    }
    if (id.indexOf('.') === -1) id = 'mqtt.0.' + id;

    mqttClient.publish(mqttid, value);

    setTimeout(() => {
        objects.getObject(id, (err, obj) => {
            expect(obj).to.be.not.null.and.not.undefined;
            expect(obj._id).to.be.equal(id);
            expect(obj.type).to.be.equal('state');

            if (mqttid.indexOf('mqtt') !== -1) {
                expect(obj.native.topic).to.be.equal(mqttid);
            }

            states.getState(id, (err, state) => {
                expect(state).to.be.not.null.and.not.undefined;
                expect(state.val).to.be.equal(value);
                expect(state.ack).to.be.true;
                _done();
            });
        });
    }, 500);
}

function checkAdapter2Mqtt(id, mqttid, _it, _done) {
    const value = 'NewRoger' + Math.round(Math.random() * 100);
    _it.timeout(5000);

    console.log('Send ' + id);

    states.setState(id, {
        val: value,
        ack: false
    }, function (err, id) {
        setTimeout(function () {
            expect(lastReceivedTopic).to.be.equal(mqttid);
            expect(lastReceivedMessage).to.be.equal(value);
            _done();
        }, 200);
    });
}

function checkConnectionOfAdapter(cb, counter) {
    counter = counter || 0;
    if (counter > 20) {
        cb && cb('Cannot check connection');
        return;
    }

    states.getState('system.adapter.mqtt.0.connected', (err, state) => {
        if (err) console.error(err);
        if (state && state.val) {
            connected = state.val;
            cb && cb();
        } else {
            setTimeout(function () {
                checkConnectionOfAdapter(cb, counter + 1);
            }, 500);
        }
    });
}

function checkConnectionToServer(value, cb, counter) {
    counter = counter || 0;
    if (counter > 20) {
        cb && cb('Cannot check connection to server for ' + value);
        return;
    }

    states.getState('mqtt.0.info.connection', (err, state) => {
        if (err) console.error(err);
        if (state && state.val == value) {
            connected = state.val;
            cb && cb();
        } else {
            setTimeout(function () {
                checkConnectionToServer(value, cb, counter + 1);
            }, 1000);
        }
    });
}

describe('Test MQTT client', function() {
    before('MQTT client: Start js-controller', function (_done) { // let FUNCTION and not => here
        this.timeout(600000); // because of first install from npm
        let clientConnected  = false;
        let brokerStarted    = false;
        setup.adapterStarted = false;

        setup.setupController(function () {
            const config = setup.getAdapterConfig();
            // enable adapter
            config.common.enabled  = true;
            config.common.loglevel = 'debug';
            config.native.publish  = 'mqtt.0.*';
            config.native.user     = 'user';
            config.native.pass     = '*\u0006\u0015\u0001\u0004';
            setup.setAdapterConfig(config.common, config.native);

            setup.startController((_objects, _states) => {
                objects = _objects;
                states  = _states;
                brokerStarted = true;
                if (_done && clientConnected) {
                    _done();
                    _done = null;
                }
            });
        });

        // start mqtt server
        MqttServer = require(__dirname + '/lib/mqttServer.js');
        const MqttClient = require(__dirname + '/lib/mqttClient.js');

        mqttServer = new MqttServer({user: 'user', pass: 'pass1'});

        // Start client to emit topics
        mqttClient = new MqttClient(() => {
            // on connected
            //console.log('Test MQTT Client is connected to MQTT broker');
            clientConnected = true;
            if (_done && brokerStarted && clientConnected) {
                _done();
                _done = null;
            }
        }, function (topic, message) {
            console.log((new Date()).getTime() + ' emitter received ' + topic + ': ' + message.toString());
            //console.log('Test MQTT Client received "' + topic + '": ' + message);
            // on receive
            lastReceivedTopic   = topic;
            lastReceivedMessage = message ? message.toString() : null;
        }, {name: 'Emitter', user: 'user', pass: 'pass1'});
    });

    it('MQTT client: Check if connected to MQTT broker', done => {
        if (!connected) {
            checkConnectionOfAdapter(done);
        } else {
            done();
        }
    }).timeout(3000);

    it('MQTT client: wait', done => {
        setTimeout(function () {
            done();
        }, 1000);
    }).timeout(4000);

    for (let rr in rules) {
        (function(id, topic) {
            it('MQTT client: Check receive ' + id, function (done) { // let FUNCTION here
                checkMqtt2Adapter(id, topic, this, done);
            });
        })(rr, rules[rr]);
    }

    for (let r in rules) {
        (function(id, topic) {
            if (topic.indexOf('mqtt') !== -1) {
                it('MQTT client: Check send ' + topic, function (done) { // let FUNCTION here
                    checkAdapter2Mqtt(topic, id, this, done);
                });
            }
        })(r, rules[r]);
    }

    it('MQTT client: check reconnect if server is down', done => {
        mqttServer.stop();
        connected = false;

        checkConnectionToServer(false, error => {
            expect(error).to.be.not.ok;
            mqttServer = new MqttServer();
            checkConnectionToServer(true, error => {
                expect(error).to.be.not.ok;
                done();
            });
        });
    }).timeout(20000);

    after('MQTT client: Stop js-controller', function (_done) { // let FUNCTION and not => here
        this.timeout(6000);
        mqttServer.stop();
        mqttClient.stop();

        setup.stopController(function (normalTerminated) {
            console.log('Adapter normal terminated: ' + normalTerminated);
            _done();
        });
    });
});