More MQTT improvements.

This commit is contained in:
Ylian Saint-Hilaire 2019-10-07 18:11:15 -07:00
parent 51b7dc4fc2
commit 241b42ac00
4 changed files with 31 additions and 21 deletions

View File

@ -2107,6 +2107,9 @@ module.exports.CreateMeshUser = function (parent, db, ws, req, args, domain, use
// Check if this user has rights to do this // Check if this user has rights to do this
if (mesh.links[user._id] != null && ((mesh.links[user._id].rights & 64) != 0)) { if (mesh.links[user._id] != null && ((mesh.links[user._id].rights & 64) != 0)) {
// If this device is connected on MQTT, send a wake action.
if (parent.parent.mqttbroker != null) { parent.parent.mqttbroker.publish(node._id, 'powerAction', 'wake'); }
// Get the device interface information // Get the device interface information
db.Get('if' + node._id, function (err, nodeifs) { db.Get('if' + node._id, function (err, nodeifs) {
if ((nodeifs != null) && (nodeifs.length == 1)) { if ((nodeifs != null) && (nodeifs.length == 1)) {
@ -2146,6 +2149,7 @@ module.exports.CreateMeshUser = function (parent, db, ws, req, args, domain, use
case 'poweraction': case 'poweraction':
{ {
if (common.validateArray(command.nodeids, 1) == false) break; // Check nodeid's if (common.validateArray(command.nodeids, 1) == false) break; // Check nodeid's
if (common.validateInt(command.actiontype, 2, 4) == false) break; // Check actiontype
for (i in command.nodeids) { for (i in command.nodeids) {
nodeid = command.nodeids[i]; nodeid = command.nodeids[i];
var powerActions = 0; var powerActions = 0;
@ -2159,6 +2163,8 @@ module.exports.CreateMeshUser = function (parent, db, ws, req, args, domain, use
// Get the mesh for this device // Get the mesh for this device
mesh = parent.meshes[node.meshid]; mesh = parent.meshes[node.meshid];
if (mesh) { if (mesh) {
// If this device is connected on MQTT, send a power action.
if (parent.parent.mqttbroker != null) { parent.parent.mqttbroker.publish(nodeid, 'powerAction', ['', '', 'poweroff', 'reset', 'sleep'][command.actiontype]); }
// Check if this user has rights to do this // Check if this user has rights to do this
if (mesh.links[user._id] != null && ((mesh.links[user._id].rights & 8) != 0)) { // "Remote Control permission" if (mesh.links[user._id] != null && ((mesh.links[user._id].rights & 8) != 0)) { // "Remote Control permission"

View File

@ -29,18 +29,9 @@ module.exports.CreateMQTTBroker = function (parent, db, args) {
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") }; 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") };
} }
// Publish a message to a specific nodeid & topic
obj.publish = function (nodeid, topic, message) {
var clients = obj.connections[nodeid];
if (clients == null) return;
if (typeof message == 'string') { message = new Buffer(message); }
for (var i in clients) { clients[i].publish({ cmd: 'publish', qos: 0, topic: topic, payload: message, retain: false }); }
}
// Connection Authentication // Connection Authentication
aedes.authenticate = function (client, username, password, callback) { aedes.authenticate = function (client, username, password, callback) {
obj.parent.debug("mqtt", "Authentication User:" + username + ", Pass:" + password.toString() + ", ClientID:" + client.id + ", " + client.conn.xtransport + "://" + cleanRemoteAddr(client.conn.xip)); obj.parent.debug("mqtt", "Authentication User:" + username + ", Pass:" + password.toString() + ", ClientID:" + client.id + ", " + client.conn.xtransport + "://" + cleanRemoteAddr(client.conn.xip));
console.log('MQTT Connect');
// Parse the username and password // Parse the username and password
var usersplit = username.split(':'); var usersplit = username.split(':');
@ -117,21 +108,30 @@ module.exports.CreateMQTTBroker = function (parent, db, args) {
// Handle a published message // Handle a published message
obj.parent.debug("mqtt", "AuthorizePublish, " + client.conn.xtransport + "://" + cleanRemoteAddr(client.conn.xip)); 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.xdbNodeKey, packet.topic, packet.payload);
//callback(denyError); // Deny all, clients can't publish anything to other agents. // We don't accept that any client message be published, so don't call the callback.
//callback(null); // Deny all, clients can't publish anything to other agents.
} }
// Check if a client can forward a packet // Publish a message to a specific nodeid & topic, also send this to peer servers.
//aedes.authorizeForward = function (client, packet) { obj.publish = function (nodeid, topic, message) {
// TODO: add forwarding control // Publish this message on peer servers.
//obj.parent.debug("mqtt", "AuthorizeForward, " + client.conn.xtransport + "://" + cleanRemoteAddr(client.conn.xip)); if (parent.multiServer != null) { parent.multiServer.DispatchMessage(JSON.stringify({ action: 'mqtt', nodeid: nodeid, topic: topic, message: message })); }
//return packet; obj.publishNoPeers(nodeid, topic, message);
//} }
// Publish a message to a specific nodeid & topic, don't send to peer servers.
obj.publishNoPeers = function (nodeid, topic, message) {
// Look for any MQTT connections to send this to
var clients = obj.connections[nodeid];
if (clients == null) return;
if (typeof message == 'string') { message = new Buffer(message); }
for (var i in clients) { clients[i].publish({ cmd: 'publish', qos: 0, topic: topic, payload: message, retain: false }); }
}
// Handle messages coming from clients // Handle messages coming from clients
function handleMessage(nodeid, meshid, topic, message) { function handleMessage(nodeid, meshid, topic, message) {
console.log('handleMessage', nodeid, topic, message.toString()); // TODO: Handle messages here.
obj.publish(nodeid, 'abc', "This is a server reply"); //console.log('handleMessage', nodeid, topic, message.toString());
//obj.publish(nodeid, 'echoTopic', "Echo: " + message.toString());
} }
// Clean a IPv6 address that encodes a IPv4 address // Clean a IPv6 address that encodes a IPv4 address

View File

@ -456,6 +456,10 @@ module.exports.CreateMultiServer = function (parent, args) {
var userid, i; var userid, i;
//console.log('ProcessPeerServerMessage', peerServerId, msg); //console.log('ProcessPeerServerMessage', peerServerId, msg);
switch (msg.action) { switch (msg.action) {
case 'mqtt': {
if ((obj.parent.mqttbroker != null) && (msg.nodeid != null)) { obj.parent.mqttbroker.publishNoPeers(msg.nodeid, msg.topic, msg.message); } // Dispatch in the MQTT broker
break;
}
case 'bus': { case 'bus': {
obj.parent.DispatchEvent(msg.ids, null, msg.event, true); // Dispatch the peer event obj.parent.DispatchEvent(msg.ids, null, msg.event, true); // Dispatch the peer event
break; break;

View File

@ -3310,7 +3310,7 @@
p10showChangeGroupDialog(getCheckedDevices()); p10showChangeGroupDialog(getCheckedDevices());
} else { } else {
// Power operation // Power operation
meshserver.send({ action: 'poweraction', nodeids: getCheckedDevices(), actiontype: op }); meshserver.send({ action: 'poweraction', nodeids: getCheckedDevices(), actiontype: parseInt(op) });
} }
} }
@ -4446,7 +4446,7 @@
meshserver.send({ action: 'wakedevices', nodeids: [ currentNode._id ] }); meshserver.send({ action: 'wakedevices', nodeids: [ currentNode._id ] });
} else { } else {
// Power operation // Power operation
meshserver.send({ action: 'poweraction', nodeids: [ currentNode._id ], actiontype: op }); meshserver.send({ action: 'poweraction', nodeids: [ currentNode._id ], actiontype: parseInt(op) });
} }
} }