'use strict'; const mqtt = require('mqtt-connection'); function Server(config) { let clients = {}; let server; let net; let http; let ws; let wsStream; let serverForWs; let serverWs; config = config || {}; function startServer(socket, server, port, bind, ssl, ws) { socket.on('connection', stream => { let client; if (ws) { client = mqtt(wsStream(stream)); } else { client = mqtt(stream); } client.on('connect', function (packet) { client.id = packet.clientId; clients[client.id] = client; if (config.user) { if (config.user !== packet.username || config.pass !== packet.password.toString()) { console.error('Client [' + packet.clientId + '] has invalid password(' + packet.password + ') or username(' + packet.username + ')'); client.connack({returnCode: 4}); if (clients[client.id]) delete clients[client.id]; client.stream.end(); return; } } console.log('Client [' + packet.clientId + '] connected: user - ' + packet.username + ', pass - ' + packet.password); client.connack({returnCode: 0}); client.publish({topic: 'testServer/connected', payload: 'true'}); }); client.on('publish', function (packet) { console.log('Client [' + client.id + '] publishes "' + packet.topic + '": ' + packet.payload); for (let k in clients) { clients[k].publish({topic: packet.topic, payload: packet.payload}); } }); client.on('subscribe', function (packet) { let granted = []; console.log('Client [' + client.id + '] subscribes on "' + JSON.stringify(packet.subscriptions) + '"'); for (let i = 0; i < packet.subscriptions.length; i++) { granted.push(packet.subscriptions[i].qos); } client.suback({granted: granted, messageId: packet.messageId}); }); client.on('pingreq', function (packet) { console.log('Client [' + client.id + '] pingreq'); client.pingresp(); }); client.on('disconnect', function (packet) { if (clients[client.id]) delete clients[client.id]; console.log('Client [' + client.id + '] disconnected'); client.stream.end(); }); client.on('close', function (err) { if (clients[client.id]) delete clients[client.id]; console.log('Client [' + client.id + '] closed'); }); client.on('error', function (err) { if (clients[client.id]) delete clients[client.id]; console.log('[' + client.id + '] ' + err); client.stream.end(); }); }); (server || socket).listen(port, bind, () => { console.log(`Starting MQTT ${ws ? '-WebSocket' : ''}${ssl ? ' (Secure)' : ''}' server on port ${port}`); }); } const port = 1883; const sslOptions = { key: "-----BEGIN RSA PRIVATE KEY-----\r\nMIICXQIBAAKBgQDQ6dVCuqpl0hdECy35tQP7n/FKAK6Yz8z04F3g8NtkLrJ3IR1+\r\nNo0ijLE2Ka5ONZV2WlRzybWomAvOGnfbSH7NG/wkQ9saBb15bAU03RLeyFmDc5Rz\r\newgjoQzJwXNWIIbqdiUWUqhy3IOzfoRrNprpDm5mv2pwEUxOuF8mB62vgQIDAQAB\r\nAoGBAKmS5DQB6IY1fgURPgROVilMrkJvQ0luguLRq+IGH062SM5B5vqntO+yW7Wn\r\nJ4D8JZGnyJ0jwXxTzmFBQsCPm7vQ3VkH1ir4JhlIWJ11Z3p3XMNWNJ5mrDAyEupn\r\nShCFQxW9EDL7efVFztqgyiWw5/uxV4AJQyBgtsF4PijmgT8xAkEA+SlmVXcuzIPy\r\nZTfNXRCWHvzZM9EaRVQXNSYqMHXLRx412gw42ihk/+GIYaw7y5ObjlMosfzzCyot\r\naMMA/KT1TwJBANalpnrDE0BhYuv/ccnxJv/pZ6aJZ4P/gyRV02UUc0WTAGnxU4el\r\nJPtREWCyCjaVq26S7fh4DGotcDhDEkpzei8CQA5aGyHrJo/zPcAk0bh9nxgT2nMI\r\npWm+6UNPenimIFptXA6+S3wNfZvbot51bFBSpVAybBKsjldjS5BQQztKSTMCQQCe\r\nMhYBkjZlE6Fhh7GogOgaYj53GfvF6BISPIMBk1HlrBL5AdhrN4aLBtOE7ZLjaemg\r\nI//pSSj1NCnp/VzErFkXAkA/6q2Th8M4Z2LzL46GeRavLXFd1IQmFULWZAkx5afk\r\n8/anbz31nnA9CFu+oR/jTp7urYsIUQ3y6ksJwGGKHVlQ\r\n-----END RSA PRIVATE KEY-----\r\n", cert: "-----BEGIN CERTIFICATE-----\r\nMIICfzCCAegCCQC1y0d8DNip4TANBgkqhkiG9w0BAQUFADCBgzELMAkGA1UEBhMC\r\nREUxGTAXBgNVBAgMEEJhZGVuV3VlcnRlbWJlcmcxEjAQBgNVBAcMCUthcmxzcnVo\r\nZTERMA8GA1UECgwIaW9Ccm9rZXIxEDAOBgNVBAMMB0JsdWVmb3gxIDAeBgkqhkiG\r\n9w0BCQEWEWRvZ2Fmb3hAZ21haWwuY29tMB4XDTE1MDQyMjIwMjgwM1oXDTE2MDQy\r\nMTIwMjgwM1owgYMxCzAJBgNVBAYTAkRFMRkwFwYDVQQIDBBCYWRlbld1ZXJ0ZW1i\r\nZXJnMRIwEAYDVQQHDAlLYXJsc3J1aGUxETAPBgNVBAoMCGlvQnJva2VyMRAwDgYD\r\nVQQDDAdCbHVlZm94MSAwHgYJKoZIhvcNAQkBFhFkb2dhZm94QGdtYWlsLmNvbTCB\r\nnzANBgkqhkiG9w0BAQEFAAOBjQAwgYkCgYEA0OnVQrqqZdIXRAst+bUD+5/xSgCu\r\nmM/M9OBd4PDbZC6ydyEdfjaNIoyxNimuTjWVdlpUc8m1qJgLzhp320h+zRv8JEPb\r\nGgW9eWwFNN0S3shZg3OUc3sII6EMycFzViCG6nYlFlKoctyDs36Eazaa6Q5uZr9q\r\ncBFMTrhfJgetr4ECAwEAATANBgkqhkiG9w0BAQUFAAOBgQBgp4dhA9HulN7/rh4H\r\n+e+hAYqjWvFpdNqwcWAyopBig9B9WL3OIkzpgTuBmH76JxzJCuZJkjO4HLGzQ3KF\r\nsFU0lvqqoz9osgYmXe1K0fBjIcm/RFazGTHVxv+UgVqQ3KldrlkvR3T2VIRlT5hI\r\n0Y1m6J3YZDMF7D6uc1jrsYHkMQ==\r\n-----END CERTIFICATE-----\r\n" }; this.start = function () { if (process.argv[2] === 'ssl') { net = net || require('tls'); if (config.webSocket) { http = http || require('https'); } } else { net = net || require('net'); if (config.webSocket) { http = http || require('http'); } } server = new net.Server(sslOptions); startServer(server, null, port, '127.0.0.1', process.argv[2] === 'ssl', false); http = http || require('https'); ws = ws || require('ws'); wsStream = wsStream || require('websocket-stream'); serverForWs = http.createServer(sslOptions); serverWs = new ws.Server({server: serverForWs}); startServer(serverWs, serverForWs, port + 1, '127.0.0.1', process.argv[2] === 'ssl', true); }; this.stop = function () { // destroy all clients (this will emit the 'close' event above) for (let i in clients) { clients[i].destroy(); } if (server) { server.close(() => { console.log('Server closed.'); server.unref(); server = null; }); } if (serverForWs) { serverForWs.close(() => { console.log('WS-Server closed.'); serverForWs.unref(); serverForWs = null; }); } }; this.start(); return this; } if (typeof module !== 'undefined' && module.parent) { module.exports = Server; } else { new Server(); }