forked from Raynerv/node-red-contrib-mqttssl
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathmqttssl.js
127 lines (114 loc) · 4.44 KB
/
mqttssl.js
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
/**
* Copyright 2013 IBM Corp.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
**/
module.exports = function(RED) {
"use strict";
var clientPool = require("./lib/mqttClientPool");
var util = require("util");
function MQTTBrokerNode(n) {
RED.nodes.createNode(this,n);
this.broker = n.broker;
this.port = n.port;
this.clientid = n.clientid;
if (this.credentials) {
this.username = this.credentials.user;
this.password = this.credentials.password;
}
}
RED.nodes.registerType("mqttssl-broker",MQTTBrokerNode,{
credentials: {
user: {type:"text"},
password: {type: "password"}
}
});
function MQTTInNode(n) {
RED.nodes.createNode(this,n);
this.topic = n.topic;
this.broker = n.broker;
this.brokerConfig = RED.nodes.getNode(this.broker);
var node = this;
if (this.brokerConfig) {
this.status({fill:"red",shape:"ring",text:"disconnected"});
this.client = clientPool.get(this.brokerConfig.broker,this.brokerConfig.port,this.brokerConfig.clientid,this.brokerConfig.username,this.brokerConfig.password);
var node = this;
this.client.subscribe(this.topic,2,function(topic,payload,pub) {
var msg = {topic:topic,payload:payload,qos:pub.qos,retain:pub.retain};
if ((node.brokerConfig.broker == "localhost")||(node.brokerConfig.broker == "127.0.0.1")) {
msg._topic = topic;
}
node.send(msg);
});
this.client.on("connectionlost",function() {
node.status({fill:"red",shape:"ring",text:"disconnected"});
});
this.client.on("connect",function() {
node.status({fill:"green",shape:"dot",text:"connected"});
});
this.client.on("error",function(error) {
node.status({fill:"red",shape:"ring",text:"error"});
node.error(error);
});
this.client.connect();
} else {
this.error("missing broker configuration");
}
}
RED.nodes.registerType("mqttssl in",MQTTInNode);
MQTTInNode.prototype.close = function() {
if (this.client) {
this.client.disconnect();
}
}
function MQTTOutNode(n) {
RED.nodes.createNode(this,n);
this.topic = n.topic;
this.broker = n.broker;
this.brokerConfig = RED.nodes.getNode(this.broker);
var node = this;
if (this.brokerConfig) {
this.status({fill:"red",shape:"ring",text:"disconnected"},true);
this.client = clientPool.get(this.brokerConfig.broker,this.brokerConfig.port,this.brokerConfig.clientid,this.brokerConfig.username,this.brokerConfig.password);
this.on("input",function(msg) {
if (msg != null) {
if (this.topic) {
msg.topic = this.topic;
}
this.client.publish(msg, function(err) {
//node.status({fill:"red",shape:"ring",text:"failed to publish"});
});
}
});
this.client.on("connectionlost",function() {
node.status({fill:"red",shape:"ring",text:"disconnected"});
});
this.client.on("error",function(error) {
node.status({fill:"red",shape:"ring",text:"error"});
node.error(error);
});
this.client.on("connect",function() {
node.status({fill:"green",shape:"dot",text:"connected"});
});
this.client.connect();
} else {
this.error("missing broker configuration");
}
}
RED.nodes.registerType("mqttssl out",MQTTOutNode);
MQTTOutNode.prototype.close = function() {
if (this.client) {
this.client.disconnect();
}
}
}