module.exports = function(RED) {
"use strict";
var StompClient = require('stomp-client');
// ----------------------------------------------
// ------------------- State --------------------
// ----------------------------------------------
function updateStatus(node, allNodes) {
let setStatus = setStatusDisconnected;
if (node.connecting) {
setStatus = setStatusConnecting;
} else if (node.connected) {
setStatus = setStatusConnected;
}
setStatus(node, allNodes);
}
function setStatusDisconnected(node, allNodes) {
if (allNodes) {
for (let id in node.users) {
if (hasProperty(node.users, id)) {
node.users[id].status({ fill: "red", shape: "ring", text: "node-red:common.status.disconnected"});
}
}
} else {
node.status({ fill: "red", shape: "ring", text: "node-red:common.status.disconnected" })
}
}
function setStatusConnecting(node, allNodes) {
if(allNodes) {
for (var id in node.users) {
if (hasProperty(node.users, id)) {
node.users[id].status({ fill: "yellow", shape: "ring", text: "node-red:common.status.connecting" });
}
}
} else {
node.status({ fill: "yellow", shape: "ring", text: "node-red:common.status.connecting" });
}
}
function setStatusConnected(node, allNodes) {
if(allNodes) {
for (var id in node.users) {
if (hasProperty(node.users, id)) {
node.users[id].status({ fill: "green", shape: "dot", text: "node-red:common.status.connected" });
}
}
} else {
node.status({ fill: "green", shape: "dot", text: "node-red:common.status.connected" });
}
}
function setStatusError(node, allNodes, message = undefined) {
if(allNodes) {
for (var id in node.users) {
if (hasProperty(node.users, id)) {
node.users[id].status({ fill: "red", shape: "dot", text: message ?? "error" });
}
}
} else {
node.status({ fill: "red", shape: "dot", text: message ?? "error" });
}
}
// ----------------------------------------------
// ------------------- Nodes --------------------
// ----------------------------------------------
function StompServerNode(n) {
RED.nodes.createNode(this,n);
const node = this;
// To keep track of processing nodes that use this config node for their connection
node.users = {};
// Config node state
node.connected = false;
node.connecting = false;
/** Flag to avoid race conditions between `deregister` and the `close` event of the config node (ex. on redeploy) */
node.closing = false;
/** Options to pass to the stomp-client API */
node.options = {};
node.sessionId = null;
node.subscribtionIndex = 1;
node.subscriptionIds = {};
/** Array of callbacks to be called once the connection to the broker has been made */
node.connectedCallbacks = [];
/** @type { StompClient } */
node.client;
node.setOptions = function(options, init) {
if (!options || typeof options !== "object") {
return; // Nothing to change
}
// Apply property changes (only if the property exists in the options object)
setIfHasProperty(options, node, "server", init);
setIfHasProperty(options, node, "port", init);
setIfHasProperty(options, node, "protocolversion", init);
setIfHasProperty(options, node, "vhost", init);
setIfHasProperty(options, node, "reconnectretries", init);
setIfHasProperty(options, node, "reconnectdelay", init);
if (node.credentials) {
node.username = node.credentials.user;
node.password = node.credentials.password;
}
if (!init && hasProperty(options, "username")) {
node.username = options.username;
}
if (!init && hasProperty(options, "password")) {
node.password = options.password;
}
// Build options for passing to the stomp-client API
node.options = {
address: node.server,
port: node.port * 1,
user: node.username,
pass: node.password,
protocolVersion: node.protocolversion,
reconnectOpts: {
retries: node.reconnectretries * 1,
delay: node.reconnectdelay * 1000
},
vhost: node.vhost
};
}
node.setOptions(n, true);
/**
* Register a STOMP processing node to the connection.
* @param { StompInNode | StompOutNode | StompAckNode } stompNode The STOMP processing node to register
* @param { Function } callback
*/
node.register = function(stompNode, callback = () => {}) {
node.users[stompNode.id] = stompNode;
if (!node.connected) {
node.connectedCallbacks.push(callback);
}
// Auto connect when first STOMP processing node is added
if (Object.keys(node.users).length === 1) {
node.connect(() => {
while (node.connectedCallbacks.length) {
node.connectedCallbacks.shift().call();
}
});
} else if (node.connected) {
// Execute callback directly as the connection to the STOMP server has already been made
callback();
}
}
/**
* Remove registered STOMP processing nodes from the connection.
* @param { StompInNode | StompOutNode | StompAckNode } stompNode The STOMP processing node to unregister
* @param { Boolean } autoDisconnect Automatically disconnect from the STOM server when no processing nodes registered to the connection
* @param { Function } callback
*/
node.deregister = function(stompNode, autoDisconnect, callback = () => {}) {
delete node.users[stompNode.id];
if (autoDisconnect && !node.closing && node.connected && Object.keys(node.users).length === 0) {
node.disconnect(callback);
} else {
callback();
}
}
/**
* Wether a new connection can be made.
* @returns `true` or `false`
*/
node.canConnect = function() {
return !node.connected && !node.connecting;
}
/**
* Connect to the STOMP server.
* @param {Function} callback
*/
node.connect = function(callback = () => {}) {
if (node.canConnect()) {
node.closing = false;
node.connecting = true;
setStatusConnecting(node, true);
try {
// Remove left over client if needed
if (node.client) {
node.client.disconnect();
node.client = null;
}
node.client = new StompClient(node.options);
node.client.on("connect", function(sessionId) {
node.closing = false;
node.connecting = false;
node.connected = true;
node.sessionId = sessionId;
node.log(`Connected to STOMP server, sessionId: ${node.sessionId}, url: ${node.options.address}:${node.options.port}, protocolVersion: ${node.options.protocolVersion}`);
setStatusConnected(node, true);
callback();
});
node.client.on("reconnect", function(sessionId, numOfRetries) {
node.closing = false;
node.connecting = false;
node.connected = true;
node.sessionId = sessionId;
node.log(`Reconnected to STOMP server, sessionId: ${node.sessionId}, url: ${node.options.address}:${node.options.port}, protocolVersion: ${node.options.protocolVersion}, retries: ${numOfRetries}`);
setStatusConnected(node, true);
callback();
});
node.client.on("reconnecting", function() {
node.connecting = true;
node.connected = false;
node.warn(`Reconnecting to STOMP server, url: ${node.options.address}:${node.options.port}, protocolVersion: ${node.options.protocolVersion}`);
setStatusConnecting(node, true);
});
node.client.on("error", function(err) {
node.error(err);
if (err.reconnectionFailed) {
setStatusError(node, true, "Reconnection failed: exceeded number of reconnection attempts");
}
});
node.client.connect();
} catch (err) {
node.error(err);
setStatusError(node, true);
}
} else {
node.log("Not connecting to STOMP server, already connected");
callback();
}
}
/**
* Disconnect from the STOMP server.
* @param {Function} callback
*/
node.disconnect = function(callback = () => {}) {
const waitDisconnect = (client, timeout) => {
return new Promise((resolve, reject) => {
// Set flag to avoid race conditions for disconnect as every node tries to call it directly or indirectly using deregister
node.closing = true;
const t = setTimeout(() => {
reject();
}, timeout);
client.disconnect(() => {
clearTimeout(t);
resolve();
});
});
}
if (!node.client) {
node.warn("Can't disconnect, connection not initialized.");
callback();
} else if (node.closing || !node.connected) {
// Disconnection already in progress or not connected
callback();
} else {
const subscribedQueues = Object.keys(node.subscriptionIds);
subscribedQueues.forEach(function(queue) {
node.unsubscribe(queue);
});
node.log('Disconnecting from STOMP server...');
waitDisconnect(node.client, 2000).then(() => {
node.log(`Disconnected from STOMP server, sessionId: ${node.sessionId}, url: ${node.options.address}:${node.options.port}, protocolVersion: ${node.options.protocolVersion}`)
}).catch(() => {
node.log("Disconnect timeout closing node...");
}).finally(() => {
node.sessionId = null;
node.subscribtionIndex = 1;
node.subscriptionIds = {};
node.connected = false;
node.connecting = false;
setStatusDisconnected(node, true);
callback();
});
}
}
/**
* Subscribe to a given STOMP queue.
* @param { String} queue The queue to subscribe to
* @param { "auto" | "client" | "client-individual" } clientAck Can be `auto`, `client` or `client-individual` (the latter only starting from STOMP v1.1)
* @param { Function } callback
*/
node.subscribe = function(queue, acknowledgment, callback) {
node.log(`Subscribe to: ${queue}`);
if (node.connected && !node.closing) {
if (!node.subscriptionIds[queue]) {
node.subscriptionIds[queue] = node.subscribtionIndex++;
}
const headers = {
id: node.subscriptionIds[queue],
// Only set client-individual if not v1.0
ack: acknowledgment === "client-individual" && node.options.protocolVersion === "1.0" ? "client" : acknowledgment
}
node.client.subscribe(queue, headers, function(body, responseHeaders) {
let msg = { headers: responseHeaders, topic: queue };
try {
msg.payload = JSON.parse(body);
} catch {
msg.payload = body;
}
callback(msg);
});
} else {
node.error("Can't subscribe, not connected");
}
}
/**
* Unsubscribe from a STOMP queue.
* @param {String} queue The STOMP queue to unsubscribe from
* @param {Object} headers Headers to add to the unsubscribe message
*/
node.unsubscribe = function(queue, headers = {}) {
delete node.subscriptionIds[queue];
if (node.connected && !node.closing) {
node.client.unsubscribe(queue, headers);
node.log(`Unsubscribed from ${queue}, headers: ${JSON.stringify(headers)}`);
}
}
/**
* Publish a STOMP message on a queue.
* @param {String} queue The STOMP queue to publish to
* @param {any} message The message to send
* @param {Object} headers STOMP headers to add to the SEND command
*/
node.publish = function(queue, message, headers = {}) {
if (node.connected && !node.closing) {
node.client.publish(queue, message, headers);
} else {
node.error("Can't publish, not connected");
}
}
/**
* Acknowledge (a) message(s) that was received from the specified queue.
* @param {String} queue The queue/topic to send an acknowledgment for
* @param {String} messageId ID of the message that was received from the server, which can be found in the reponse header as `message-id`
* @param {String} transaction Optional transaction name
*/
node.ack = function(queue, messageId, transaction = undefined) {
if (node.connected && !node.closing) {
node.client.ack(messageId, node.subscriptionIds[queue], transaction);
} else {
node.error("Can't send acknowledgment, not connected");
}
}
node.on("close", function(done) {
node.disconnect(function() { done (); });
});
}
RED.nodes.registerType("stomp-server", StompServerNode, {
credentials: {
user: { type: "text" },
password: { type: "password" }
}
});
function StompInNode(n) {
RED.nodes.createNode(this,n);
/** @type { StompInNode } */
const node = this;
node.server = n.server;
/** @type { StompServerNode } */
node.serverConnection = RED.nodes.getNode(node.server);
node.topic = n.topic;
node.ack = n.ack;
if (node.serverConnection) {
setStatusDisconnected(node);
if (node.topic) {
node.serverConnection.register(node, function() {
node.serverConnection.subscribe(node.topic, node.ack, function(msg) {
node.send(msg);
});
});
if (node.serverConnection.connected) {
setStatusConnected(node);
}
}
} else {
node.error("Missing server config");
}
node.on("close", function(removed, done) {
if (node.serverConnection) {
node.serverConnection.unsubscribe(node.topic);
node.serverConnection.deregister(node, true, done);
node.serverConnection = null;
} else {
done();
}
});
}
RED.nodes.registerType("stomp in",StompInNode);
function StompOutNode(n) {
RED.nodes.createNode(this,n);
/** @type { StompOutNode } */
const node = this;
node.server = n.server;
/** @type { StompServerNode } */
node.serverConnection = RED.nodes.getNode(node.server);
node.topic = n.topic;
if (node.serverConnection) {
setStatusDisconnected(node);
node.on("input", function(msg, send, done) {
const topic = node.topic || msg.topic;
if (topic.length > 0 && msg.payload) {
try {
msg.payload = JSON.stringify(msg.payload);
} catch {
msg.payload = `${msg.payload}`;
}
node.serverConnection.publish(topic, msg.payload, msg.headers || {});
} else if (!topic.length > 0) {
node.warn('No valid publish topic');
} else {
node.warn('Payload or topic is undefined/null')
}
done();
});
node.serverConnection.register(node);
if (node.serverConnection.connected) {
setStatusConnected(node);
}
} else {
node.error("Missing server config");
}
node.on("close", function(removed, done) {
if (node.serverConnection) {
node.serverConnection.deregister(node, true, done);
node.serverConnection = null;
} else {
done();
}
});
}
RED.nodes.registerType("stomp out",StompOutNode);
function StompAckNode(n) {
RED.nodes.createNode(this,n);
/** @type { StompOutNode } */
const node = this;
node.server = n.server;
/** @type { StompServerNode } */
node.serverConnection = RED.nodes.getNode(node.server);
node.topic = n.topic;
if (node.serverConnection) {
setStatusDisconnected(node);
node.on("input", function(msg, send, done) {
const topic = node.topic || msg.topic;
if (topic.length > 0) {
node.serverConnection.ack(topic, msg.messageId, msg.transaction);
} else if (!topic.length > 0) {
node.warn('No valid publish topic');
} else {
node.warn('Payload or topic is undefined/null')
}
done();
});
node.serverConnection.register(node);
if (node.serverConnection.connected) {
setStatusConnected(node);
}
} else {
node.error("Missing server config");
}
node.on("close", function(removed, done) {
if (node.serverConnection) {
node.serverConnection.deregister(node, true, done);
node.serverConnection = null;
} else {
done();
}
});
}
RED.nodes.registerType("stomp ack",StompAckNode);
};
// ----------------------------------------------
// ----------------- Helpers --------------------
// ----------------------------------------------
/**
* Helper function for applying changes to an objects properties ONLY when the src object actually has the property.
* This avoids setting a `dst` property null/undefined when the `src` object doesnt have the named property.
* @param {object} src Source object containing properties
* @param {object} dst Destination object to set property
* @param {string} propName The property name to set in the Destination object
* @param {boolean} force force the dst property to be updated/created even if src property is empty
*/
function setIfHasProperty(src, dst, propName, force) {
if (src && dst && propName) {
const ok = force || hasProperty(src, propName);
if (ok) {
dst[propName] = src[propName];
}
}
}
/**
* Helper function to test an object has a property
* @param {object} obj Object to test
* @param {string} propName Name of property to find
* @returns true if object has property `propName`
*/
function hasProperty(obj, propName) {
//JavaScript does not protect the property name hasOwnProperty
//Object.prototype.hasOwnProperty.call is the recommended/safer test
return Object.prototype.hasOwnProperty.call(obj, propName);
}