More MQTT improvements.
Ylian Saint-Hilaire committed
Oct 7, 2019 at 18:11 UTC
d5f7943684584cd26ad52f6fc1ec98f3f7c226e7
4 files changed
+31
-21
meshuser.js
+6
@@ -2107,6 +2107,9 @@ module.exports.CreateMeshUser = function (parent, db, ws, req, args, domain, use
2107
// Check if this user has rights to do this
2108
if (mesh.links[user._id] != null && ((mesh.links[user._id].rights & 64) != 0)) {
2109
2110
+ // If this device is connected on MQTT, send a wake action.
2111
+ if (parent.parent.mqttbroker != null) { parent.parent.mqttbroker.publish(node._id, 'powerAction', 'wake'); }
2112
+
2113
// Get the device interface information
2114
db.Get('if' + node._id, function (err, nodeifs) {
2115
if ((nodeifs != null) && (nodeifs.length == 1)) {
@@ -2146,6 +2149,7 @@ module.exports.CreateMeshUser = function (parent, db, ws, req, args, domain, use
2149
case 'poweraction':
2150
{
2151
if (common.validateArray(command.nodeids, 1) == false) break; // Check nodeid's
2152
+ if (common.validateInt(command.actiontype, 2, 4) == false) break; // Check actiontype
2153
for (i in command.nodeids) {
2154
nodeid = command.nodeids[i];
2155
var powerActions = 0;
@@ -2159,6 +2163,8 @@ module.exports.CreateMeshUser = function (parent, db, ws, req, args, domain, use
2163
// Get the mesh for this device
2164
mesh = parent.meshes[node.meshid];
2165
if (mesh) {
2166
+ // If this device is connected on MQTT, send a power action.
2167
+ if (parent.parent.mqttbroker != null) { parent.parent.mqttbroker.publish(nodeid, 'powerAction', ['', '', 'poweroff', 'reset', 'sleep'][command.actiontype]); }
2168
2169
// Check if this user has rights to do this
2170
if (mesh.links[user._id] != null && ((mesh.links[user._id].rights & 8) != 0)) { // "Remote Control permission"
mqttbroker.js
+19
-19
@@ -29,18 +29,9 @@ module.exports.CreateMQTTBroker = function (parent, db, args) {
29
return { meshid: meshid, nodeid: nodeid, user: username, pass: parent.config.settings.mqtt.auth.keyid + ':' + nonce + ':' + parent.crypto.createHash('sha384').update(username + ':' + nonce + ':' + parent.config.settings.mqtt.auth.key).digest("base64") };
30
}
31
32
- // Publish a message to a specific nodeid & topic
33
- obj.publish = function (nodeid, topic, message) {
34
- var clients = obj.connections[nodeid];
35
- if (clients == null) return;
36
- if (typeof message == 'string') { message = new Buffer(message); }
37
- for (var i in clients) { clients[i].publish({ cmd: 'publish', qos: 0, topic: topic, payload: message, retain: false }); }
38
- }
39
-
32
// Connection Authentication
33
aedes.authenticate = function (client, username, password, callback) {
34
obj.parent.debug("mqtt", "Authentication User:" + username + ", Pass:" + password.toString() + ", ClientID:" + client.id + ", " + client.conn.xtransport + "://" + cleanRemoteAddr(client.conn.xip));
43
- console.log('MQTT Connect');
35
36
// Parse the username and password
37
var usersplit = username.split(':');
@@ -117,21 +108,30 @@ module.exports.CreateMQTTBroker = function (parent, db, args) {
108
// Handle a published message
109
obj.parent.debug("mqtt", "AuthorizePublish, " + client.conn.xtransport + "://" + cleanRemoteAddr(client.conn.xip));
110
handleMessage(client.xdbNodeKey, client.xdbNodeKey, packet.topic, packet.payload);
120
- //callback(denyError); // Deny all, clients can't publish anything to other agents.
121
- //callback(null); // Deny all, clients can't publish anything to other agents.
111
+ // We don't accept that any client message be published, so don't call the callback.
112
}
113
124
- // Check if a client can forward a packet
125
- //aedes.authorizeForward = function (client, packet) {
126
- // TODO: add forwarding control
127
- //obj.parent.debug("mqtt", "AuthorizeForward, " + client.conn.xtransport + "://" + cleanRemoteAddr(client.conn.xip));
128
- //return packet;
129
- //}
114
+ // Publish a message to a specific nodeid & topic, also send this to peer servers.
115
+ obj.publish = function (nodeid, topic, message) {
116
+ // Publish this message on peer servers.
117
+ if (parent.multiServer != null) { parent.multiServer.DispatchMessage(JSON.stringify({ action: 'mqtt', nodeid: nodeid, topic: topic, message: message })); }
118
+ obj.publishNoPeers(nodeid, topic, message);
119
+ }
120
+
121
+ // Publish a message to a specific nodeid & topic, don't send to peer servers.
122
+ obj.publishNoPeers = function (nodeid, topic, message) {
123
+ // Look for any MQTT connections to send this to
124
+ var clients = obj.connections[nodeid];
125
+ if (clients == null) return;
126
+ if (typeof message == 'string') { message = new Buffer(message); }
127
+ for (var i in clients) { clients[i].publish({ cmd: 'publish', qos: 0, topic: topic, payload: message, retain: false }); }
128
+ }
129
130
// Handle messages coming from clients
131
function handleMessage(nodeid, meshid, topic, message) {
133
- console.log('handleMessage', nodeid, topic, message.toString());
134
- obj.publish(nodeid, 'abc', "This is a server reply");
132
+ // TODO: Handle messages here.
133
+ //console.log('handleMessage', nodeid, topic, message.toString());
134
+ //obj.publish(nodeid, 'echoTopic', "Echo: " + message.toString());
135
}
136
137
// Clean a IPv6 address that encodes a IPv4 address
multiserver.js
+4
@@ -456,6 +456,10 @@ module.exports.CreateMultiServer = function (parent, args) {
456
var userid, i;
457
//console.log('ProcessPeerServerMessage', peerServerId, msg);
458
switch (msg.action) {
459
+ case 'mqtt': {
460
+ if ((obj.parent.mqttbroker != null) && (msg.nodeid != null)) { obj.parent.mqttbroker.publishNoPeers(msg.nodeid, msg.topic, msg.message); } // Dispatch in the MQTT broker
461
+ break;
462
+ }
463
case 'bus': {
464
obj.parent.DispatchEvent(msg.ids, null, msg.event, true); // Dispatch the peer event
465
break;
views/default.handlebars
+2
-2
@@ -3310,7 +3310,7 @@
3310
p10showChangeGroupDialog(getCheckedDevices());
3311
} else {
3312
// Power operation
3313
- meshserver.send({ action: 'poweraction', nodeids: getCheckedDevices(), actiontype: op });
3313
+ meshserver.send({ action: 'poweraction', nodeids: getCheckedDevices(), actiontype: parseInt(op) });
3314
}
3315
}
3316
@@ -4446,7 +4446,7 @@
4446
meshserver.send({ action: 'wakedevices', nodeids: [ currentNode._id ] });
4447
} else {
4448
// Power operation
4449
- meshserver.send({ action: 'poweraction', nodeids: [ currentNode._id ], actiontype: op });
4449
+ meshserver.send({ action: 'poweraction', nodeids: [ currentNode._id ], actiontype: parseInt(op) });
4450
}
4451
}
4452