'use strict'; const expect = require('chai').expect; const Adapter = require('./lib/adapterSim'); const Server = require('../lib/server'); const Client = require('./lib/mqttClient'); let port = 1883; describe('MQTT server', () => { let adapter; let server; let states = {}; before('MQTT server: Start MQTT server', done => { adapter = new Adapter({ port: ++port, defaultQoS: 1, onchange: true }); server = new Server(adapter, states); done(); }); it('MQTT server: Check if connected to MQTT broker', done => { let client = new Client(isConnected => { if (done) { expect(isConnected).to.be.true; client.destroy(); done(); done = null; } }, null, { url: 'localhost:' + port, clientId: 'testClient1', } ); }); it('MQTT server: Check if subscribes stored', () => { let client; const data = 1; return new Promise(resolve => { client = new Client(isConnected => { if (isConnected) { client.subscribe('aaa'); setTimeout(() => client.destroy(), 200); // let time to send it out } else { adapter.setForeignState('mqtt.0.aaa', data); server.onStateChange('mqtt.0.aaa', {val: data, ack: false}); setTimeout(() => resolve(), 100); } }, null, { url: 'localhost:' + port, clean: false, clientId: 'testClient2', resubscribe: false } ); }) .then(() => { new Promise(resolve => { client = new Client( () => { }, (topic, message) => { if (topic === 'aaa') { expect(topic).to.be.equal('aaa'); expect(message.toString()).to.be.equal(data.toString()); client.destroy(); resolve(); } }, { url: 'localhost:' + port, clean: false, clientId: 'testClient2', resubscribe: false } ); }) }); }); it('MQTT server: Check if QoS1 retransmitted', done => { let client; const data = 1; let sendPacket; let count = 0; const id = 'aaa2'; let allowPuback = false; let receiveFunc; new Promise(resolve => { client = new Client(isConnected => { if (isConnected) { client.subscribe(id, {qos: 1}); setTimeout(() => resolve(), 100); } }, (topic, data) => receiveFunc && receiveFunc(topic, data), { url: 'localhost:' + port, clean: false, clientId: 'testClient3', resubscribe: false } ); sendPacket = client.client._sendPacket; client.client._sendPacket = function (packet, cb) { // ignore puback if (packet.cmd === 'puback' && !allowPuback) { count++; cb && cb(); return; } sendPacket.call(this, packet, cb); }; }) .then(() => { return new Promise(resolve => { adapter.setForeignState('mqtt.0.' + id, data); server.onStateChange('mqtt.0.' + id, {val: data, ack: false}); setTimeout(() => resolve(), 1000); }); }) .then(() => { console.log(`[${new Date().toISOString()} continue tests`); expect(count).to.be.equal(1); allowPuback = true; receiveFunc = () => { client.destroy(); done(); }; }); }).timeout(5000); it('MQTT server: Check if QoS2 retransmitted', done => { let receiverClient; let emitterClient; const data = 1; const id = 'aaa3'; let sendPacket; let count = 0; let allowPubrec = false; let receiveFunc; new Promise(resolve => { receiverClient = new Client(isConnected => { if (isConnected) { receiverClient.subscribe(id, {qos: 2}); setTimeout(() => resolve(), 100); } }, (topic, data) => receiveFunc && receiveFunc(topic, data), { url: 'localhost:' + port, clean: false, clientId: 'receiverClient', resubscribe: false } ); emitterClient = new Client(null, null, { url: 'localhost:' + port, clean: true, clientId: 'emitterClient', resubscribe: false } ); sendPacket = receiverClient.client._sendPacket; receiverClient.client._sendPacket = function (packet, cb) { // ignore pubrec if (packet.cmd === 'pubrec' && !allowPubrec) { count++; cb && cb(); return; } sendPacket.call(this, packet, cb); }; }) .then(() => { return new Promise(resolve => { emitterClient.publish(id, data.toString(), 2); // Send QoS 2 setTimeout(() => resolve(), 100); }); }) .then(() => { expect(count).to.be.equal(1); allowPubrec = true; receiveFunc = () => { receiverClient.destroy(); emitterClient.destroy(); done(); }; }); }).timeout(5000); it('MQTT server: Check if message with QoS1 received', done => { let receiverClient; let emitterClient; const data = 1; const id = 'aaa4'; let receiveFunc; new Promise(resolve => { receiverClient = new Client(isConnected => { if (isConnected) { receiverClient.subscribe(id, {qos: 1}); setTimeout(() => resolve(), 100); } }, (topic, data, packet) => receiveFunc && receiveFunc(topic, data, packet), { url: 'localhost:' + port, clean: false, clientId: 'receiverClient', resubscribe: false } ); emitterClient = new Client(null, null, { url: 'localhost:' + port, clean: true, clientId: 'emitterClient', resubscribe: false } ); }) .then(() => { return new Promise(resolve => { receiveFunc = (topic, data, packet) => { expect(data).to.be.ok; expect(topic).to.be.ok; expect(packet.qos).to.be.equal(1); receiverClient.destroy(); emitterClient.destroy(); done(); }; emitterClient.publish(id, data.toString(), 1); // Send QoS 2 setTimeout(() => resolve(), 100); }); }); }).timeout(1000); // check unsubscribe it('MQTT server: Check if unsubscribes works', () => { let client; const data = 1; let count = 0; return new Promise(resolve => { client = new Client(isConnected => { if (isConnected) { client.subscribe('aaa6'); setTimeout(() => { adapter.setForeignState('mqtt.0.aaa6', data); server.onStateChange('mqtt.0.aaa6', {val: data, ack: false}); }, 500); } }, (id, topic, packet) => { if (id.indexOf('aaa6') !== -1) { console.log('Received ' + topic.toString()); count++; expect(count).to.be.equal(1); setTimeout(() => resolve(), 100); } }, { url: 'localhost:' + port, clean: true, clientId: 'testClient6', resubscribe: false } ); }) .then(() => { return new Promise(resolve => { client.unsubscribe('aaa6'); client.unsubscribe('#'); setTimeout(() => { console.log('Resend data'); adapter.setForeignState('mqtt.0.aaa6', 2); server.onStateChange('mqtt.0.aaa6', {val: 2, ack: false}); // wait 1 second to not receive the update setTimeout(() => { console.log('Done'); client.destroy(); resolve(); }, 1000); }, 300); }); }); }).timeout(3000); after('MQTT server: Stop MQTT server', done => { server.destroy(done); }); });