frodovdr/node-red-nodes
1
0
Fork 0
mirror of https://github.com/node-red/node-red-nodes.git synced 2025-03-01 10:37:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

Improvements regarding cleanup on close

Browse Source
This commit is contained in:
Olivier Verhaegen 2023-04-17 15:09:54 +02:00 committed by GitHub
parent ed74a69b58
commit c3628b6d91
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 10 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
io/stomp/18-stomp.js
View File

@ -256,6 +256,11 @@ module.exports = function(RED) {
// Disconnection already in progress or not connected // Disconnection already in progress or not connected
callback(); callback();
} else { } else {
node.log("Unsubscribing from STOMP queue's...");
const subscribedQueues = Object.keys(node.subscriptionIds);
subscribedQueues.forEach(function(queue) {
node.unsubscribe(queue);
});
node.log('Disconnecting from STOMP server...'); node.log('Disconnecting from STOMP server...');
waitDisconnect(node.client, 2000).then(() => { 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}) node.log("Disconnected from STOMP server", {sessionId: node.sessionId, url: `${node.options.address}:${node.options.port}`, protocolVersion: node.options.protocolVersion})
@ -282,7 +287,7 @@ module.exports = function(RED) {
node.subscribe = function(queue, acknowledgement, callback) { node.subscribe = function(queue, acknowledgement, callback) {
node.log(`Subscribing to: ${queue}`); node.log(`Subscribing to: ${queue}`);
if (node.connected) { if (node.connected && !node.closing) {
if (!node.subscriptionIds[queue]) { if (!node.subscriptionIds[queue]) {
node.subscriptionIds[queue] = node.subscribtionIndex++; node.subscriptionIds[queue] = node.subscribtionIndex++;
} }
@ -314,7 +319,7 @@ module.exports = function(RED) {
*/ */
node.unsubscribe = function(queue, headers = {}) { node.unsubscribe = function(queue, headers = {}) {
delete node.subscriptionIds[queue]; delete node.subscriptionIds[queue];
if (node.connected) { if (node.connected && !node.closing) {
node.client.unsubscribe(queue, headers); node.client.unsubscribe(queue, headers);
node.log(`Unsubscribed from ${queue}`, headers); node.log(`Unsubscribed from ${queue}`, headers);
} }
@ -327,7 +332,7 @@ module.exports = function(RED) {
* @param {Object} headers STOMP headers to add to the SEND command * @param {Object} headers STOMP headers to add to the SEND command
*/ */
node.publish = function(queue, message, headers = {}) { node.publish = function(queue, message, headers = {}) {
if (node.connected) { if (node.connected && !node.closing) {
node.client.publish(queue, message, headers); node.client.publish(queue, message, headers);
} else { } else {
node.error("Can't publish, not connected"); node.error("Can't publish, not connected");
@ -341,7 +346,7 @@ module.exports = function(RED) {
* @param {String} transaction Optional transaction name * @param {String} transaction Optional transaction name
*/ */
node.ack = function(queue, messageId, transaction = undefined) { node.ack = function(queue, messageId, transaction = undefined) {
if (node.connected) { if (node.connected && !node.closing) {
node.client.ack(messageId, node.subscriptionIds[queue], transaction); node.client.ack(messageId, node.subscriptionIds[queue], transaction);
} else { } else {
node.error("Can't send acknowledgement, not connected"); node.error("Can't send acknowledgement, not connected");
@ -389,9 +394,7 @@ module.exports = function(RED) {
node.on("close", function(removed, done) { node.on("close", function(removed, done) {
if (node.serverConnection) { if (node.serverConnection) {
if (node.serverConnection.connected) { node.serverConnection.unsubscribe(node.topic);
node.serverConnection.unsubscribe(node.topic);
}
node.serverConnection.deregister(node, true, done); node.serverConnection.deregister(node, true, done);
node.serverConnection = null; node.serverConnection = null;
} else { } else {