实现[email protected] 系列的监控

PomeloClient/Client.js

@@ -0,0 +1,160 @@
+var MqttClient = require('./mqttClient');
+var protocol = require('./protocol');
+var crypto = require('crypto');
+
+var Client = function(opt) {
+    this.id = "";
+    this.reqId = 1;
+    this.callbacks = {};
+    this.listeners = {};
+    this.state = Client.ST_INITED;
+    this.socket = null;
+    opt = opt || {};
+    this.username = opt['username'] || "";
+    this.password = opt['password'] || "";
+    this.md5 = opt['md5'] || false;
+};
+
+Client.prototype = {
+    connect: function(id, host, port, cb) {
+        this.id = id;
+        var self = this;
+
+        console.log('try to connect ' + host + ':' + port);
+        this.socket = new MqttClient({
+            id: id
+        });
+
+        this.socket.connect(host, port);
+
+        // this.socket = io.connect('http://' + host + ':' + port, {
+        // 	'force new connection': true,
+        // 	'reconnect': false
+        // });
+
+        this.socket.on('connect', function() {
+            self.state = Client.ST_CONNECTED;
+            if (self.md5) {
+                self.password = md5(self.password);
+            }
+            self.doSend('register', {
+                type: "client",
+                id: id,
+                username: self.username,
+                password: self.password,
+                md5: self.md5
+            });
+        });
+
+        this.socket.on('register', function(res) {
+            if (res.code !== protocol.PRO_OK) {
+                cb(res.msg);
+                return;
+            }
+
+            self.state = Client.ST_REGISTERED;
+            cb();
+        });
+
+        this.socket.on('client', function(msg) {
+            msg = protocol.parse(msg);
+            if (msg.respId) {
+                // response for request
+                var cb = self.callbacks[msg.respId];
+                delete self.callbacks[msg.respId];
+                if (cb && typeof cb === 'function') {
+                    cb(msg.error, msg.body);
+                }
+            } else if (msg.moduleId) {
+                // notify
+                self.emit(msg.moduleId, msg);
+            }
+        });
+
+        this.socket.on('error', function(err) {
+            if (self.state < Client.ST_CONNECTED) {
+                cb(err);
+            }
+
+            self.emit('error', err);
+        });
+
+        this.socket.on('disconnect', function(reason) {
+            this.state = Client.ST_CLOSED;
+            self.emit('close');
+        });
+    },
+
+    request: function(moduleId, msg, cb) {
+        var id = this.reqId++;
+        // something dirty: attach current client id into msg
+        msg = msg || {};
+        msg.clientId = this.id;
+        msg.username = this.username;
+        var req = protocol.composeRequest(id, moduleId, msg);
+        this.callbacks[id] = cb;
+        this.doSend('client', req);
+        // this.socket.emit('client', req);
+    },
+
+    notify: function(moduleId, msg) {
+        // something dirty: attach current client id into msg
+        msg = msg || {};
+        msg.clientId = this.id;
+        msg.username = this.username;
+        var req = protocol.composeRequest(null, moduleId, msg);
+        this.doSend('client', req);
+        // this.socket.emit('client', req);
+    },
+
+    command: function(command, moduleId, msg, cb) {
+        var id = this.reqId++;
+        msg = msg || {};
+        msg.clientId = this.id;
+        msg.username = this.username;
+        var commandReq = protocol.composeCommand(id, command, moduleId, msg);
+        this.callbacks[id] = cb;
+        this.doSend('client', commandReq);
+        // this.socket.emit('client', commandReq);
+    },
+
+    doSend: function(topic, msg) {
+        this.socket.send(topic, msg);
+    },
+
+    on: function(event, listener) {
+        this.listeners[event] = this.listeners[event] || [];
+        this.listeners[event].push(listener);
+    },
+
+    emit: function(event) {
+        var listeners = this.listeners[event];
+        if (!listeners || !listeners.length) {
+            return;
+        }
+
+        var args = Array.prototype.slice.call(arguments, 1);
+        var listener;
+        for (var i = 0, l = listeners.length; i < l; i++) {
+            listener = listeners[i];
+            if (typeof listener === 'function') {
+                listener.apply(null, args);
+            }
+        }
+    }
+};
+
+function md5(str)
+{
+    var md5sum = crypto.createHash('md5');
+    md5sum.update(str);
+    str = md5sum.digest('hex');
+    return str;
+}
+
+Client.ST_INITED = 1;
+Client.ST_CONNECTED = 2;
+Client.ST_REGISTERED = 3;
+Client.ST_CLOSED = 4;
+
+module.exports = Client;

PomeloClient/WebServerRoute.js

@@ -0,0 +1,106 @@
+const net = require('net');
+const MqttCon = require('mqtt-connection');
+const config = require('./../config/admin');
+const websocket = require('websocket-stream');
+const protcol = require('./protocol');
+const WebSocketServer = require('ws').Server;
+const adminClient = require('./Client');
+
+var client = null;
+function WebServer()
+{
+    const wss = new WebSocketServer({port: config.webPort});
+    wss.on('connection', function (ws) {
+        const stream = websocket(ws);
+        const socket = MqttCon(stream);
+
+        socket.on('connect', function(pkg) {
+            socket.connack({
+                returnCode: 0
+            });
+        });
+
+        socket.on('publish', function(pkg)
+        {
+            const topic = pkg.topic;
+            var msg = pkg.payload.toString();
+            msg = JSON.parse(msg);
+            msg = protcol.parse(msg);
+            if (topic === 'register')
+            {
+                msg['host'] = config.host;
+                msg['port'] = config.port;
+                connectToMaster(msg.id, msg);
+            }
+            else
+            {
+                if (client === null)
+                {
+                    socket.removeAllListeners();
+                    socket.disconnect();
+                    socket.destroy();
+                    return;
+                }
+                (function (msg, topic, socket)
+                {
+                    const moduleId = msg.moduleId;
+                    const body = msg.body;
+                    var command = body.command;
+                    if (command)
+                    {
+                        command = {command:command};
+                    }
+                    client.request(moduleId, command, (err,data) =>
+                    {
+                        if (data)
+                        {
+                            const payload = protcol.composeResponse(msg, err, data);
+                            if (payload)
+                            {
+                                socket.publish({
+                                    topic: topic,
+                                    payload: payload
+                                });
+                            }
+
+                        }
+                        else
+                        {
+                            console.info(msg);
+                        }
+                    })
+                })(msg, topic, socket);
+            }
+        });
+
+        socket.on('pingreq', function() {
+            socket.pingresp();
+        });
+    })
+}
+
+function connectToMaster(id, opts) {
+    client = new adminClient({username: opts.username, password: opts.password, md5: opts.md5});
+    client.connect(id, opts.host, opts.port,function(err)
+    {
+        if(err) {
+            client = null;
+            console.error(err);
+            process.exit(1);
+        }
+    });
+    client.on('error', function ()
+    {
+        client = null;
+    });
+    client.on('close', function ()
+    {
+        client = null;
+    });
+    client.on('disconnect', function()
+    {
+        client = null;
+    });
+}
+
+module.exports = WebServer;