From 54ec71f16c2210ca7f8e8c5a8630c9c06fd85836 Mon Sep 17 00:00:00 2001
From: Olivier Verhaegen <56387556+OlivierVerhaegen@users.noreply.github.com>
Date: Tue, 11 Apr 2023 09:32:36 +0200
Subject: [PATCH] Add support for ACK messages

---
 io/stomp/18-stomp.html | 58 ++++++++++++++++++++++++++++++++++
 io/stomp/18-stomp.js   | 72 +++++++++++++++++++++++++++++++++++++++++-
 2 files changed, 129 insertions(+), 1 deletion(-)

diff --git a/io/stomp/18-stomp.html b/io/stomp/18-stomp.html
index e643807c..2d652604 100644
--- a/io/stomp/18-stomp.html
+++ b/io/stomp/18-stomp.html
@@ -122,6 +122,14 @@
             <option value="1.1">v1.1</option>
         </select>
     </div>
+    <div class="form-row">
+        <label for="node-config-input-ack"><i class="fa fa-check"></i> Enable client acknowledgement</label>
+        <input type="checkbox" id="node-config-input-ack" />
+    </div>
+    <div class="form-tips">
+        Enabling the ACK (acknowledgement) will set the <code>ack</code> header to <code>client</code> while subscribing to topics. 
+        This means the items on the broker queue will not be dequeue'd unless an ACK message is sent using the <code>ack</code> node.
+    </div>
     <div class="form-row">
         <label for="node-config-input-vhost"><i class="fa fa-server"></i> vhost</label>
         <input type="text" id="node-config-input-vhost" placeholder="Default is null" />
@@ -147,6 +155,7 @@
             server: {required:true},
             port: {value:61613,required:true,validate:RED.validators.number()},
             protocolversion: {value:"1.0",required:true},
+            ack: {value: false},
             vhost: {},
             reconnectretries: {value:"0",required:true,validate:RED.validators.number()},
             reconnectdelay: {value:"0.5",required:true,validate:RED.validators.number()},
@@ -161,3 +170,52 @@
         }
     });
 </script>
+
+<script type="text/html" data-template-name="stomp ack">
+    <div class="form-row">
+        <label for="node-input-server" style="width: 110px;"><i class="fa fa-bookmark"></i> Server</label>
+        <input type="text" id="node-input-server">
+    </div>
+    <div class="form-row">
+        <label for="node-input-name" style="width: 110px;"><i class="fa fa-tag"></i> Name</label>
+        <input type="text" id="node-input-name" placeholder="Name">
+    </div>
+</script>
+
+<script type="text/html" data-help-name="stomp ack">
+    <p>
+        ACK is used to acknowledge consumption of a message from a subscription using client acknowledgment. When a client has issued a SUBSCRIBE frame with the ack header set to client any messages received from that destination will not be considered to have been consumed (by the server) until the message has been acknowledged via an ACK.
+    </p>
+    <p>
+        The node allows for following inputs:
+        <ul>
+            <li><code>msg.messageId</code>: The id of the message to acknowledge</li>
+            <li><code>msg.subscription</code>: The id of the subscription made using the stomp in node</li>
+            <li><code>msg.transaction</code>: Optional transaction name</li>
+        </ul>
+    </p>
+    <p>
+        See the <a href="https://stomp.github.io/stomp-specification-1.0.html#frame-ACK" target="new">Stomp 1.0 spec</a> for more details.
+    </p>
+</script>
+
+<script type="text/javascript">
+    RED.nodes.registerType('stomp ack',{
+        category: 'output',
+        color:"#e8cfe8",
+        defaults: {
+            name: {value:""},
+            server: {type:"stomp-server",required:true}
+        },
+        inputs:1,
+        outputs:0,
+        icon: "bridge.png",
+        align: "right",
+        label: function() {
+            return this.name||"stomp";
+        },
+        labelStyle: function() {
+            return (this.name)?"node_label_italic":"";
+        }
+    });
+</script>
diff --git a/io/stomp/18-stomp.js b/io/stomp/18-stomp.js
index a35da1a7..b44b605c 100644
--- a/io/stomp/18-stomp.js
+++ b/io/stomp/18-stomp.js
@@ -9,6 +9,7 @@ module.exports = function(RED) {
         this.server = n.server;
         this.port = n.port;
         this.protocolversion = n.protocolversion;
+        this.ack = n.ack;
         this.vhost = n.vhost;
         this.reconnectretries = n.reconnectretries || 999999;
         this.reconnectdelay = (n.reconnectdelay || 15) * 1000;
@@ -43,6 +44,7 @@ module.exports = function(RED) {
         if (this.serverConfig.vhost) {
             this.stompClientOpts.vhost = this.serverConfig.vhost;
         }
+        this.subscribeHeaders = this.serverConfig.ack ? { "ack": "client" } : {};
 
         var node = this;
         var msg = {topic:this.topic};
@@ -69,7 +71,7 @@ module.exports = function(RED) {
         node.status({fill:"grey",shape:"ring",text:"connecting"});
         node.client.connect(function(sessionId) {
             node.log('subscribing to: '+node.topic);
-            node.client.subscribe(node.topic, function(body, headers) {
+            node.client.subscribe(node.topic, node.subscribeHeaders, function(body, headers) {
                 var newmsg={"headers":headers,"topic":node.topic}
                 try {
                     newmsg.payload = JSON.parse(body);
@@ -158,4 +160,72 @@ module.exports = function(RED) {
     }
     RED.nodes.registerType("stomp out",StompOutNode);
 
+    function StompAckNode(n) {
+        RED.nodes.createNode(this,n);
+        this.server = n.server;
+
+        this.serverConfig = RED.nodes.getNode(this.server);
+        this.stompClientOpts = {
+            address: this.serverConfig.server,
+            port: this.serverConfig.port * 1,
+            user: this.serverConfig.username,
+            pass: this.serverConfig.password,
+            protocolVersion: this.serverConfig.protocolversion,
+            reconnectOpts: {
+                retries: this.serverConfig.reconnectretries * 1,
+                delay: this.serverConfig.reconnectdelay * 1
+            }
+        };
+        if (this.serverConfig.vhost) {
+            this.stompClientOpts.vhost = this.serverConfig.vhost;
+        }
+
+        var node = this;
+
+        // only start connection etc. when acknowledgements are configured to be send by client
+        if (node.serverConfig.ack) {
+            node.client = new StompClient(node.stompClientOpts);
+
+            node.client.on("connect", function() {
+                node.status({fill:"green",shape:"dot",text:"connected"});
+            });
+
+            node.client.on("reconnecting", function() {
+                node.status({fill:"red",shape:"ring",text:"reconnecting"});
+                node.warn("reconnecting");
+            });
+
+            node.client.on("reconnect", function() {
+                node.status({fill:"green",shape:"dot",text:"connected"});
+            });
+
+            node.client.on("error", function(error) {
+                node.status({fill:"grey",shape:"dot",text:"error"});
+                node.warn(error);
+            });
+
+            node.status({fill:"grey",shape:"ring",text:"connecting"});
+            node.client.connect(function(sessionId) {
+            }, function(error) {
+                node.status({fill:"grey",shape:"dot",text:"error"});
+                node.warn(error);
+            });
+
+            node.on("input", function(msg) {
+                node.ack(msg.messageId, msg.subsriptionId, msg.transaction);
+            });
+
+            node.on("close", function(done) {
+                if (node.client) {
+                    // disconnect can accept a callback - but it is not always called.
+                    node.client.disconnect();
+                }
+                done();
+            });
+        } else {
+            node.error("ACK not configured in server (config node)");
+        }
+    }
+    RED.nodes.registerType("stomp ack",StompAckNode);
+
 };