More MQTT improvements.

mqttbroker.js

@@ -107,7 +107,7 @@ module.exports.CreateMQTTBroker = function (parent, db, args) {
     aedes.authorizePublish = function (client, packet, callback) {
         // Handle a published message
         obj.parent.debug("mqtt", "AuthorizePublish, " + client.conn.xtransport + "://" + cleanRemoteAddr(client.conn.xip));
-        handleMessage(client.xdbNodeKey, client.xdbNodeKey, packet.topic, packet.payload);
+        handleMessage(client.xdbNodeKey, client.xdbMeshKey, packet.topic, packet.payload);
         // We don't accept that any client message be published, so don't call the callback.
     }
 
@@ -129,7 +129,9 @@ module.exports.CreateMQTTBroker = function (parent, db, args) {
 
     // Handle messages coming from clients
     function handleMessage(nodeid, meshid, topic, message) {
-        // TODO: Handle messages here.
+        // Handle messages here
+        if (topic == 'console') { routeMessage({ action: 'msg', type: 'console', value: message.toString(), source: 'MQTT' }, nodeid, meshid); return; } // Handle console messages
+
         //console.log('handleMessage', nodeid, topic, message.toString());
         //obj.publish(nodeid, 'echoTopic', "Echo: " + message.toString());
     }
@@ -137,5 +139,79 @@ module.exports.CreateMQTTBroker = function (parent, db, args) {
     // Clean a IPv6 address that encodes a IPv4 address
     function cleanRemoteAddr(addr) { if (typeof addr != 'string') { return null; } if (addr.indexOf('::ffff:') == 0) { return addr.substring(7); } else { return addr; } }
 
+    // Route a message
+    function routeMessage(command, dbNodeKey, dbMeshKey) {
+        // Route a message.
+        // If this command has a sessionid, that is the target.
+        if (command.sessionid != null) {
+            if (typeof command.sessionid != 'string') return;
+            var splitsessionid = command.sessionid.split('/');
+            // Check that we are in the same domain and the user has rights over this node.
+            if ((splitsessionid[0] == 'user') && (splitsessionid[1] == domain.id)) {
+                // Check if this user has rights to get this message
+                //if (mesh.links[user._id] == null || ((mesh.links[user._id].rights & 16) == 0)) return; // TODO!!!!!!!!!!!!!!!!!!!!!
+
+                // See if the session is connected. If so, go ahead and send this message to the target node
+                var ws = parent.webserver.wssessions2[command.sessionid];
+                if (ws != null) {
+                    command.nodeid = dbNodeKey; // Set the nodeid, required for responses.
+                    delete command.sessionid;       // Remove the sessionid, since we are sending to that sessionid, so it's implyed.
+                    try { ws.send(JSON.stringify(command)); } catch (ex) { }
+                } else if (parent.multiServer != null) {
+                    // See if we can send this to a peer server
+                    var serverid = parent.webserver.wsPeerSessions2[command.sessionid];
+                    if (serverid != null) {
+                        command.fromNodeid = dbNodeKey;
+                        parent.multiServer.DispatchMessageSingleServer(command, serverid);
+                    }
+                }
+            }
+        } else if (command.userid != null) { // If this command has a userid, that is the target.
+            if (typeof command.userid != 'string') return;
+            var splituserid = command.userid.split('/');
+            // Check that we are in the same domain and the user has rights over this node.
+            if ((splituserid[0] == 'user') && (splituserid[1] == domain.id)) {
+                // Check if this user has rights to get this message
+                //if (mesh.links[user._id] == null || ((mesh.links[user._id].rights & 16) == 0)) return; // TODO!!!!!!!!!!!!!!!!!!!!!
+
+                // See if the session is connected
+                var sessions = parent.webserver.wssessions[command.userid];
+
+                // Go ahead and send this message to the target node
+                if (sessions != null) {
+                    command.nodeid = dbNodeKey; // Set the nodeid, required for responses.
+                    delete command.userid;          // Remove the userid, since we are sending to that userid, so it's implyed.
+                    for (i in sessions) { sessions[i].send(JSON.stringify(command)); }
+                }
+
+                if (parent.multiServer != null) {
+                    // TODO: Add multi-server support
+                }
+            }
+        } else { // Route this command to the mesh
+            command.nodeid = dbNodeKey;
+            var cmdstr = JSON.stringify(command);
+            for (var userid in parent.webserver.wssessions) { // Find all connected users for this mesh and send the message
+                var user = parent.webserver.users[userid];
+                if ((user != null) && (user.links != null)) {
+                    var rights = user.links[dbMeshKey];
+                    if (rights != null) { // TODO: Look at what rights are needed for message routing
+                        var xsessions = parent.webserver.wssessions[userid];
+                        // Send the message to all users on this server
+                        for (i in xsessions) { try { xsessions[i].send(cmdstr); } catch (e) { } }
+                    }
+                }
+            }
+
+            // Send the message to all users of other servers
+            if (parent.multiServer != null) {
+                delete command.nodeid;
+                command.fromNodeid = dbNodeKey;
+                command.meshid = dbMeshKey;
+                parent.multiServer.DispatchMessage(command);
+            }
+        }
+    }
+
     return obj;
 }