From 906703db5f6134e5dc99670b8f2e7d143e894938 Mon Sep 17 00:00:00 2001
From: Dave Conway-Jones <conway@uk.ibm.com>
Date: Sun, 20 Mar 2016 17:46:12 +0000
Subject: [PATCH] Add timed release mode to delay node

---
 nodes/core/core/89-delay.html          | 25 +++++++++++++---
 nodes/core/core/89-delay.js            | 13 ++++++--
 nodes/core/locales/en-US/messages.json |  4 ++-
 test/nodes/core/core/89-delay_spec.js  | 41 ++++++++++++++++++++++++++
 4 files changed, 75 insertions(+), 8 deletions(-)

diff --git a/nodes/core/core/89-delay.html b/nodes/core/core/89-delay.html
index 73104e5ab..937727c76 100644
--- a/nodes/core/core/89-delay.html
+++ b/nodes/core/core/89-delay.html
@@ -19,10 +19,11 @@
     <div class="form-row">
         <label for="node-input-pauseType"><i class="fa fa-tasks"></i> <span data-i18n="delay.action"></span></label>
         <select id="node-input-pauseType" style="width:270px !important">
-          <option value="delay" data-i18n="delay.delaymsg"></option>
-          <option value="random" data-i18n="delay.randomdelay"></option>
-          <option value="rate" data-i18n="delay.limitrate"></option>
-          <option value="queue" data-i18n="delay.fairqueue"></option>
+            <option value="delay" data-i18n="delay.delaymsg"></option>
+            <option value="random" data-i18n="delay.randomdelay"></option>
+            <option value="rate" data-i18n="delay.limitrate"></option>
+            <option value="queue" data-i18n="delay.fairqueue"></option>
+            <option value="timed" data-i18n="delay.timedqueue"></option>
         </select>
     </div>
     <div id="delay-details" class="form-row">
@@ -82,6 +83,9 @@
     At each "tick", derived from the rate, the next "topic" is released.
     Any messages arriving on the same topic before release replace those in that position in the queue.
     So each "topic" gets a turn - but the most recent value is always the one sent.</p>
+    <p>The "timed release queue" adds messages to an array based on their <code>msg.topic</code> property.
+    At each "tick", all the latest messages are released. Any new messages arriving before release will
+    replace those of the same topic.</p>
 </script>
 
 <script type="text/javascript">
@@ -114,6 +118,9 @@
             } else if (this.pauseType == "random") {
               return this.name || this._("delay.label.random");
             }
+            else if (this.pauseType == "timed") {
+                return this.name || this.rate+" "+this._("delay.label.timed")+" "+this.rateUnits;
+            }
             else {
                 var units = this.rateUnits ? this.rateUnits.charAt(0) : "s";
                 return this.name || this._("delay.label.queue")+" "+this.rate+" msg/"+units;
@@ -149,6 +156,11 @@
             $("#rate-details").show();
             $("#random-details").hide();
             $("#node-input-dr").hide();
+          } else if (this.pauseType == "timed") {
+            $("#delay-details").hide();
+            $("#rate-details").show();
+            $("#random-details").hide();
+            $("#node-input-dr").hide();
           }
 
           if (!this.timeoutUnits) {
@@ -184,6 +196,11 @@
               $("#rate-details").show();
               $("#random-details").hide();
               $("#node-input-dr").hide();
+            } else if (this.value == "timed") {
+              $("#delay-details").hide();
+              $("#rate-details").show();
+              $("#random-details").hide();
+              $("#node-input-dr").hide();
             }
           });
         }
diff --git a/nodes/core/core/89-delay.js b/nodes/core/core/89-delay.js
index 47f9c01a8..2132dccc0 100644
--- a/nodes/core/core/89-delay.js
+++ b/nodes/core/core/89-delay.js
@@ -147,10 +147,17 @@ module.exports = function(RED) {
                 node.status({});
             });
 
-        } else if (this.pauseType === "queue") {
+        } else if ((this.pauseType === "queue") || (this.pauseType === "timed")) {
             this.intervalID = setInterval(function() {
-                if (node.buffer.length > 0) {
-                    node.send(node.buffer.shift()); // send the first on the queue
+                if (this.pauseType === "queue") {
+                    if (node.buffer.length > 0) {
+                        node.send(node.buffer.shift()); // send the first on the queue
+                    }
+                }
+                else {
+                    while (node.buffer.length > 0) {    // send the whole queue
+                        node.send(node.buffer.shift());
+                    }
                 }
                 node.status({text:node.buffer.length});
             },node.rate);
diff --git a/nodes/core/locales/en-US/messages.json b/nodes/core/locales/en-US/messages.json
index b911f1985..09c93bf9f 100644
--- a/nodes/core/locales/en-US/messages.json
+++ b/nodes/core/locales/en-US/messages.json
@@ -152,6 +152,7 @@
         "randomdelay": "Random delay",
         "limitrate": "Limit rate to",
         "fairqueue": "Topic based fair queue",
+        "timedqueue": "Timed release queue",
         "milisecs": "Miliseconds",
         "secs": "Seconds",
         "sec": "Second",
@@ -169,7 +170,8 @@
             "delay": "delay",
             "limit": "limit",
             "random": "random",
-            "queue": "queue"
+            "queue": "queue",
+            "timed": "releases per"
         },
         "error": {
             "buffer": "buffer exceeded 1000 messages"
diff --git a/test/nodes/core/core/89-delay_spec.js b/test/nodes/core/core/89-delay_spec.js
index d5cb1f27c..fa91cfb80 100644
--- a/test/nodes/core/core/89-delay_spec.js
+++ b/test/nodes/core/core/89-delay_spec.js
@@ -477,4 +477,45 @@ describe('delay Node', function() {
         });
     });
 
+    it('handles timed queue', function(done) {
+        this.timeout(6000);
+
+        var flow = [{"id":"delayNode1","type":"delay","name":"delayNode","pauseType":"timed","timeout":5,"timeoutUnits":"seconds","rate":1000,"rateUnits":"second","randomFirst":"1","randomLast":"5","randomUnits":"seconds","drop":false,"wires":[["helperNode1"]]},
+                    {id:"helperNode1", type:"helper", wires:[]}];
+        helper.load(delayNode, flow, function() {
+            var delayNode1 = helper.getNode("delayNode1");
+            var helperNode1 = helper.getNode("helperNode1");
+            var messages = 2;
+            var c = 0;
+
+            helperNode1.on("input", function(msg) {
+                c += 1;
+                if (c === 1) {
+                    msg.should.have.property("topic","A");
+                    msg.should.have.property("payload",3);
+                }
+                else if (c === 2) {
+                    msg.should.have.property("topic","_none_");
+                    msg.should.have.property("payload",2);
+                }
+                else if (c === 3) {
+                    msg.should.have.property("topic","_none_");
+                    msg.should.have.property("payload","Biscuit");
+                    done();
+                }
+            });
+
+            // send test messages
+            delayNode1.receive({payload:1,topic:"A"});
+            delayNode1.receive({payload:1});
+            delayNode1.receive({payload:2,topic:"A"});
+            delayNode1.receive({payload:3,topic:"A"});      // should get this
+            delayNode1.receive({payload:2});                // and this
+            setTimeout( function() {
+                delayNode1.receive({payload:"Biscuit"});    // and then this
+            },2000);
+
+        });
+    });
+
 });