# How to process any messaging queues synchronously in node-red

**URL:** <https://discourse.nodered.org/t/how-to-process-any-messaging-queues-synchronously-in-node-red/71805>\
**Category:** General\
**Created:** [7 December 2022 06:13 UTC](https://discourse.nodered.org/t/how-to-process-any-messaging-queues-synchronously-in-node-red/71805 "2022-12-07T06:13:08Z")\
**Posts on this page:** 13\
**Page:** 1

<div class="post-metadata">

**Author:** ![shubham-credence](https://sea2.discourse-cdn.com/flex026/user_avatar/discourse.nodered.org/shubham-credence/32/69376_2.png) [@shubham-credence](https://discourse.nodered.org/u/shubham-credence)\
**Post date:** [7 December 2022 06:13 UTC](https://discourse.nodered.org/t/how-to-process-any-messaging-queues-synchronously-in-node-red/71805/1 "2022-12-07T06:13:08Z")

</div>

Hi,  
I'm using `Node-RED version: v1.3.7` and `Node.js version: v14.17.5` .  
Is their any packages or node-red core `nodes` exists in node-red which process AQMP messages synchronously.

Because i am facing issues, while processing large amount of RabbitMQ messages, so i wanted to process all the messages synchronously.

Thanks in advanced for any help!

---

<div class="post-metadata">

**Author:** ![Colin](https://sea2.discourse-cdn.com/flex026/user_avatar/discourse.nodered.org/colin/32/17040_2.png) [@Colin](https://discourse.nodered.org/u/Colin)\
**Post date:** [7 December 2022 07:48 UTC](https://discourse.nodered.org/t/how-to-process-any-messaging-queues-synchronously-in-node-red/71805/2 "2022-12-07T07:48:23Z")

</div>

Can you explain what you mean by that please?

---

<div class="post-metadata">

**Author:** ![shubham-credence](https://sea2.discourse-cdn.com/flex026/user_avatar/discourse.nodered.org/shubham-credence/32/69376_2.png) [@shubham-credence](https://discourse.nodered.org/u/shubham-credence)\
**Post date:** [7 December 2022 10:50 UTC](https://discourse.nodered.org/t/how-to-process-any-messaging-queues-synchronously-in-node-red/71805/3 "2022-12-07T10:50:05Z")

</div>

What i mean is:  
For e.g. I have 200 messages to process in one of the custom node, so the node is processing all the messages at the same time, so i wanted a node-red `nodes` or packages which send my each messages to custom node synchronously( one after the other).

Is their any predefined node in node-red or any package similar to package: `node-red-contrib-amqp`.  
While useing the `node-red-contrib-amqp` package its requires the `node.js version: between version 6 to version 8` and my current `node.js version is v14.17.5`.

---

<div class="post-metadata">

**Author:** ![Steve-Mcl](https://sea2.discourse-cdn.com/flex026/user_avatar/discourse.nodered.org/steve-mcl/32/4826_2.png) [@Steve-Mcl](https://discourse.nodered.org/u/Steve-Mcl)\
**Post date:** [7 December 2022 11:00 UTC](https://discourse.nodered.org/t/how-to-process-any-messaging-queues-synchronously-in-node-red/71805/4 "2022-12-07T11:00:11Z")

</div>

> [@shubham-credence](#):
>
> i wanted a node-red `nodes` or packages which send my each messages to custom node synchronously( one after the other)

Have you tried the `split` node?

what does the data coming out of `node-red-contrib-amqp` look like?

- Is it an array of 200 things?
- is it an object with 200 keys?
- Is it sending out 200 separate messages?

---

<div class="post-metadata">

**Author:** ![Colin](https://sea2.discourse-cdn.com/flex026/user_avatar/discourse.nodered.org/colin/32/17040_2.png) [@Colin](https://discourse.nodered.org/u/Colin)\
**Post date:** [7 December 2022 11:03 UTC](https://discourse.nodered.org/t/how-to-process-any-messaging-queues-synchronously-in-node-red/71805/5 "2022-12-07T11:03:52Z")

</div>

> [@shubham-credence](#):
>
> I have 200 messages to process in one of the custom node, so the node is processing all the messages at the same time,

Do you mean 200 messages, or 200 sets of data all in one message? Node red nodes all process just one message at a time, they are not processed at the same time.

---

<div class="post-metadata">

**Author:** ![shubham-credence](https://sea2.discourse-cdn.com/flex026/user_avatar/discourse.nodered.org/shubham-credence/32/69376_2.png) [@shubham-credence](https://discourse.nodered.org/u/shubham-credence)\
**Post date:** [7 December 2022 12:01 UTC](https://discourse.nodered.org/t/how-to-process-any-messaging-queues-synchronously-in-node-red/71805/6 "2022-12-07T12:01:25Z")

</div>

Hi @Colin

I mean 200 messages, and all process of messages should be in synchronously.

Issues: currently i am facing issues is, while process 200 messages all at one time, some messages does not get process they got skiped.  
Result i want: if with the help of some packages/library or nodes of nodered flows, i want them to process only one after the other.

---

<div class="post-metadata">

**Author:** ![Colin](https://sea2.discourse-cdn.com/flex026/user_avatar/discourse.nodered.org/colin/32/17040_2.png) [@Colin](https://discourse.nodered.org/u/Colin)\
**Post date:** [7 December 2022 12:14 UTC](https://discourse.nodered.org/t/how-to-process-any-messaging-queues-synchronously-in-node-red/71805/7 "2022-12-07T12:14:55Z")

</div>

If you mean that you have a section of flow and you want to wait till one message has completely passed through that section before the next one is allowed through then have a look at this example of how to do it. Set the time in the Queue node to a greater time than your flow should ever take to run through. If something locks up then it will release the next message after that time in order to let things keep going.

 ![image](https://us1.discourse-cdn.com/flex026/uploads/nodered/original/3X/4/9/496d6567b805b1ced6476a4407ef3b9af11391f1.png)

```auto
[{"id":"b6630ded2db7d680","type":"inject","z":"bdd7be38.d3b55","name":"","props":[{"p":"payload"},{"p":"topic","vt":"str"}],"repeat":"","crontab":"","once":false,"onceDelay":0.1,"topic":"","payload":"","payloadType":"date","x":120,"y":1700,"wires":[["ed63ee4225312b40"]]},{"id":"ed63ee4225312b40","type":"delay","z":"bdd7be38.d3b55","name":"Queue","pauseType":"rate","timeout":"5","timeoutUnits":"seconds","rate":"1","nbRateUnits":"1","rateUnits":"minute","randomFirst":"1","randomLast":"5","randomUnits":"seconds","drop":false,"allowrate":false,"outputs":1,"x":290,"y":1700,"wires":[["d4d479e614e82a49","7eb760e019b512dc"]]},{"id":"a82c03c3d34f683c","type":"delay","z":"bdd7be38.d3b55","name":"Some more stuff to do","pauseType":"delay","timeout":"5","timeoutUnits":"seconds","rate":"1","nbRateUnits":"1","rateUnits":"second","randomFirst":"1","randomLast":"5","randomUnits":"seconds","drop":false,"allowrate":false,"outputs":1,"x":780,"y":1700,"wires":[["7c6253e5d34769ac","b23cea1074943d4d"]]},{"id":"2128a855234c1016","type":"link in","z":"bdd7be38.d3b55","name":"link in 1","links":["7c6253e5d34769ac"],"x":75,"y":1780,"wires":[["3a9faf0a95b4a9bb"]]},{"id":"7c6253e5d34769ac","type":"link out","z":"bdd7be38.d3b55","name":"link out 1","mode":"link","links":["2128a855234c1016"],"x":645,"y":1780,"wires":[]},{"id":"b23cea1074943d4d","type":"debug","z":"bdd7be38.d3b55","name":"OUT","active":true,"tosidebar":true,"console":false,"tostatus":false,"complete":"payload","targetType":"msg","statusVal":"","statusType":"auto","x":650,"y":1620,"wires":[]},{"id":"d4d479e614e82a49","type":"debug","z":"bdd7be38.d3b55","name":"IN","active":true,"tosidebar":true,"console":false,"tostatus":false,"complete":"payload","targetType":"msg","statusVal":"","statusType":"auto","x":450,"y":1620,"wires":[]},{"id":"3a9faf0a95b4a9bb","type":"function","z":"bdd7be38.d3b55","name":"Flush","func":"return {flush: 1}","outputs":1,"noerr":0,"initialize":"","finalize":"","libs":[],"x":170,"y":1780,"wires":[["ed63ee4225312b40"]]},{"id":"7eb760e019b512dc","type":"function","z":"bdd7be38.d3b55","name":"Some functions to be performed","func":"\nreturn msg;","outputs":1,"noerr":0,"initialize":"","finalize":"","libs":[],"x":530,"y":1700,"wires":[["a82c03c3d34f683c"]]}]

```

---

<div class="post-metadata">

**Author:** ![marcus-j-davies](https://sea2.discourse-cdn.com/flex026/user_avatar/discourse.nodered.org/marcus-j-davies/32/103435_2.png) [@marcus-j-davies](https://discourse.nodered.org/u/marcus-j-davies)\
**Post date:** [7 December 2022 15:25 UTC](https://discourse.nodered.org/t/how-to-process-any-messaging-queues-synchronously-in-node-red/71805/8 "2022-12-07T15:25:06Z")

</div>

Not sure I am on the right track, but....

Another interesting approach (and one I use)  
is using `node-red-contrib-semaphore`

> **[node-red-contrib-semaphore](https://flows.nodered.org/node/node-red-contrib-semaphore)**
>
> A Node-RED node for semaphores.

its not perfect, but does allow to signal when another message can pass through the network.  
setting the token bucket to 1, will effectively only allow 1 message at a time. That is when you signal the last message has been processed

---

<div class="post-metadata">

**Author:** ![Colin](https://sea2.discourse-cdn.com/flex026/user_avatar/discourse.nodered.org/colin/32/17040_2.png) [@Colin](https://discourse.nodered.org/u/Colin)\
**Post date:** [7 December 2022 16:14 UTC](https://discourse.nodered.org/t/how-to-process-any-messaging-queues-synchronously-in-node-red/71805/9 "2022-12-07T16:14:19Z")

</div>

The example I posted does effectively the same as the semaphore nodes with two advantages, firstly it uses just core nodes, and secondly it has a built in recovery option if something goes wrong. It is easy to get the semaphore nodes locked up. For example a redeploy while there is a message in progress can cause it. I used to use the semaphore nodes but I have moved all mine across to use the technique I posted.

---

<div class="post-metadata">

**Author:** ![E1cid](https://sea2.discourse-cdn.com/flex026/user_avatar/discourse.nodered.org/e1cid/32/77971_2.png) [@E1cid](https://discourse.nodered.org/u/E1cid)\
**Post date:** [7 December 2022 16:27 UTC](https://discourse.nodered.org/t/how-to-process-any-messaging-queues-synchronously-in-node-red/71805/10 "2022-12-07T16:27:07Z")

</div>

Was the flush 1 in the rate limit delay node introduced after Node-red 1.3.7? as the OP is on 1.3.7. I seem to recall it was V2

---

<div class="post-metadata">

**Author:** ![marcus-j-davies](https://sea2.discourse-cdn.com/flex026/user_avatar/discourse.nodered.org/marcus-j-davies/32/103435_2.png) [@marcus-j-davies](https://discourse.nodered.org/u/marcus-j-davies)\
**Post date:** [7 December 2022 16:31 UTC](https://discourse.nodered.org/t/how-to-process-any-messaging-queues-synchronously-in-node-red/71805/11 "2022-12-07T16:31:08Z")

</div>

Nice!  
I may steel this for myself!

But I agree, the semaphore node can be temperamental at times!

---

<div class="post-metadata">

**Author:** ![Colin](https://sea2.discourse-cdn.com/flex026/user_avatar/discourse.nodered.org/colin/32/17040_2.png) [@Colin](https://discourse.nodered.org/u/Colin)\
**Post date:** [7 December 2022 16:41 UTC](https://discourse.nodered.org/t/how-to-process-any-messaging-queues-synchronously-in-node-red/71805/12 "2022-12-07T16:41:36Z")

</div>

> [@E1cid](#):
>
> Was the flush 1 in the rate limit delay node introduced after Node-red 1.3.7

I don't know, in so he/she will soon find out. My test flow can be imported and tested to see if it works. I think the example has a 5 second delay in the 'do something' section so it should let one message through every 5 seconds.

@shubham-credence can you not upgrade? That is a very old version.

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/flex026/uploads/nodered/original/1X/d073cd938eafa2e558d7c2cd59003b3ef4963033.png) [@system](https://discourse.nodered.org/u/system)\
**Post date:** [5 February 2023 16:42 UTC](https://discourse.nodered.org/t/how-to-process-any-messaging-queues-synchronously-in-node-red/71805/13 "2023-02-05T16:42:29Z")

</div>

This topic was automatically closed 60 days after the last reply. New replies are no longer allowed.