dgbuilder/core_nodes/core/89-delay.js

   1 /**
   2  * Copyright 2013, 2014 IBM Corp.
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at
   7  *
   8  * http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  **/
  16
  17 //Simple node to introduce a pause into a flow
  18 module.exports = function(RED) {
  19     "use strict";
  20
  21     var MILLIS_TO_NANOS = 1000000;
  22     var SECONDS_TO_NANOS = 1000000000;
  23
  24     function random(n) {
  25         var wait = n.randomFirst + (n.diff * Math.random());
  26         if (n.buffer.length > 0) {
  27             n.send(n.buffer.pop());
  28             n.randomID = setTimeout(function() {random(n);},wait);
  29         } else {
  30             n.randomID = -1;
  31         }
  32     }
  33
  34     function DelayNode(n) {
  35         RED.nodes.createNode(this,n);
  36
  37         this.pauseType = n.pauseType;
  38         this.timeoutUnits = n.timeoutUnits;
  39         this.randomUnits = n.randomUnits;
  40         this.rateUnits = n.rateUnits;
  41
  42         if (n.timeoutUnits === "milliseconds") {
  43             this.timeout = n.timeout;
  44         } else if (n.timeoutUnits === "seconds") {
  45             this.timeout = n.timeout * 1000;
  46         } else if (n.timeoutUnits === "minutes") {
  47             this.timeout = n.timeout * (60 * 1000);
  48         } else if (n.timeoutUnits === "hours") {
  49             this.timeout = n.timeout * (60 * 60 * 1000);
  50         } else if (n.timeoutUnits === "days") {
  51             this.timeout = n.timeout * (24 * 60 * 60 * 1000);
  52         }
  53
  54         if (n.rateUnits === "second") {
  55             this.rate = 1000/n.rate;
  56         } else if (n.rateUnits === "minute") {
  57             this.rate = (60 * 1000)/n.rate;
  58         } else if (n.rateUnits === "hour") {
  59             this.rate = (60 * 60 * 1000)/n.rate;
  60         } else if (n.rateUnits === "day") {
  61             this.rate = (24 * 60 * 60 * 1000)/n.rate;
  62         }
  63
  64         if (n.randomUnits === "milliseconds") {
  65             this.randomFirst = n.randomFirst;
  66             this.randomLast = n.randomLast;
  67         } else if (n.randomUnits === "seconds") {
  68             this.randomFirst = n.randomFirst * 1000;
  69             this.randomLast = n.randomLast * 1000;
  70         } else if (n.randomUnits === "minutes") {
  71             this.randomFirst = n.randomFirst * (60 * 1000);
  72             this.randomLast = n.randomLast * (60 * 1000);
  73         } else if (n.randomUnits === "hours") {
  74             this.randomFirst = n.randomFirst * (60 * 60 * 1000);
  75             this.randomLast = n.randomLast * (60 * 60 * 1000);
  76         } else if (n.randomUnits === "days") {
  77             this.randomFirst = n.randomFirst * (24 * 60 * 60 * 1000);
  78             this.randomLast = n.randomLast * (24 * 60 * 60 * 1000);
  79         }
  80
  81         this.diff = this.randomLast - this.randomFirst;
  82         this.name = n.name;
  83         this.idList = [];
  84         this.buffer = [];
  85         this.intervalID = -1;
  86         this.randomID = -1;
  87         this.lastSent;
  88         this.drop = n.drop;
  89         var node = this;
  90
  91         if (this.pauseType === "delay") {
  92             this.on("input", function(msg) {
  93                 var id;
  94                 id = setTimeout(function(){
  95                     node.idList.splice(node.idList.indexOf(id),1);
  96                     node.send(msg);
  97                 }, node.timeout);
  98                 this.idList.push(id);
  99             });
 100
 101             this.on("close", function() {
 102                 for (var i=0; i<this.idList.length; i++ ) {
 103                     clearTimeout(this.idList[i]);
 104                 }
 105                 this.idList = [];
 106             });
 107
 108         } else if (this.pauseType === "rate") {
 109             this.on("input", function(msg) {
 110                 if (!node.drop) {
 111                     if ( node.intervalID !== -1) {
 112                         node.buffer.push(msg);
 113                         if (node.buffer.length > 0) {
 114                             node.status({text:node.buffer.length});
 115                         }
 116                         if (node.buffer.length > 1000) {
 117                             node.warn(this.name + " buffer exceeded 1000 messages");
 118                         }
 119                     } else {
 120                         node.send(msg);
 121                         node.intervalID = setInterval(function() {
 122                             if (node.buffer.length === 0) {
 123                                 clearInterval(node.intervalID);
 124                                 node.intervalID = -1;
 125                                 node.status({text:""});
 126                             }
 127
 128                             if (node.buffer.length > 0) {
 129                                 node.send(node.buffer.shift());
 130                                 node.status({text:node.buffer.length});
 131                             }
 132                         },node.rate);
 133                     }
 134                 } else {
 135                     var timeSinceLast;
 136                     if (node.lastSent) {
 137                         timeSinceLast = process.hrtime(node.lastSent);
 138                     }
 139                     if (!node.lastSent) { // ensuring that we always send the first message
 140                         node.lastSent = process.hrtime();
 141                         node.send(msg);
 142                     } else if ( ( (timeSinceLast[0] * SECONDS_TO_NANOS) + timeSinceLast[1] ) > (node.rate * MILLIS_TO_NANOS) ) {
 143                         node.lastSent = process.hrtime();
 144                         node.send(msg);
 145                     }
 146                 }
 147             });
 148
 149             this.on("close", function() {
 150                 clearInterval(this.intervalID);
 151                 this.buffer = [];
 152             });
 153
 154         } else if (this.pauseType === "random") {
 155             this.on("input",function(msg){
 156                 node.buffer.push(msg);
 157                 if (node.randomID === -1) {
 158                     var wait = node.randomFirst + (node.diff * Math.random());
 159                     node.randomID = setTimeout(function() {random(node);},wait);
 160                 }
 161             });
 162
 163             this.on("close", function (){
 164                 if (this.randomID !== -1) {
 165                     clearTimeout(this.randomID);
 166                 }
 167             });
 168         }
 169     }
 170     RED.nodes.registerType("delay",DelayNode);
 171 }