node-red-contrib-msg-queue-plus/queue-plus.js at master · damianbenente/node-red-contrib-msg-queue-plus · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
module.exports = function (RED) {

    function QueuePlus(config) {
        RED.nodes.createNode(this, config);
        this.autoTriggerOn = config.autoTriggerOn;
        this.autoTriggerTime = config.autoTriggerTime;
        this.sendComplete = config.sendComplete;
        this.queueFullMessage = config.queueFullMessage;
        var node = this;
        var context = this.context();
        context.queue = context.queue || [];
        context.busy = context.busy || false;


        node.on('input', function (msg) {
            node.send(process(msg));
            updateStatus();
        });

        function process(msg) {
            if (msg.hasOwnProperty("reset")) {
                context.queue = [];
                context.busy = false;
            } else if (msg.hasOwnProperty("trigger")) {
                if (context.queue.length > 0) {
                    var m = context.queue.shift();
                    var out1 = {};
                    if (node.queueFullMessage === true) {
                        out1 = m;
                    } else {
                        out1.payload = m;
                    }
                    if (context.queue.length === 0) {
                        if (node.sendComplete === true) out1.complete = true;
                        node.send([null, { payload: 'end' }]);
                        context.busy = false;
                    }
                    return [out1, null, {
                        payload: context.queue.length
                    }];
                } else {
                    context.busy = false;
                    return [null, {
                        payload: 'nodata'
                    }];
                }
            } else {

                if (node.queueFullMessage === true) {
                    context.queue.push(msg);
                } else {
                    context.queue.push(msg.payload);
                }
                var x = context.busy;
                context.busy = true;
                if (x === false && node.autoTriggerOn === true) {
                    sleep(node.autoTriggerTime).then(() => {
                        //node.log("triggered autotriger");

                        node.send([null, {
                            payload: 'start'
                        }]);
                        msg.trigger = true;
                        node.send(process(msg));
                    })
                } else {
                    if (x === false) node.send([null, {payload: 'dataLoadedWaiting'}]);
                    return [null, null, {
                        payload: context.queue.length
                    }];
                }
            }
        }

        var updateStatus = throttle(function () {
            if (context.busy) {
                node.status({
                    fill: "blue",
                    shape: "dot",
                    text: context.queue.length
                });
            } else {
                node.status({
                    fill: "blue",
                    shape: "ring",
                    text: context.queue.length
                });
            }
        }, 1000);

        function sleep(time) {
            return new Promise((resolve) => setTimeout(resolve, time));
        }

        function throttle(func, wait, immediate) {
            var timeout;
            return function () {
                var context = this,
                    args = arguments;
                var later = function () {
                    timeout = null;
                    if (!immediate) func.apply(context, args);
                };
                var callNow = immediate && !timeout;
                if (!timeout) timeout = setTimeout(later, wait);
                if (callNow) func.apply(context, args);
            };
        };

    }

    RED.nodes.registerType("queue-plus", QueuePlus);
}