# Creating an async node that uses pub/sub

**URL:** <https://discourse.nodered.org/t/creating-an-async-node-that-uses-pub-sub/90787>\
**Category:** General\
**Tags:** mqtt, async, kafka\
**Created:** [8 September 2024 08:50 UTC](https://discourse.nodered.org/t/creating-an-async-node-that-uses-pub-sub/90787 "2024-09-08T08:50:39Z")\
**Posts on this page:** 6\
**Page:** 1

<div class="post-metadata">

**Author:** ![seb](https://sea2.discourse-cdn.com/flex026/user_avatar/discourse.nodered.org/seb/32/94157_2.png) [@seb](https://discourse.nodered.org/u/seb)\
**Post date:** [8 September 2024 08:50 UTC](https://discourse.nodered.org/t/creating-an-async-node-that-uses-pub-sub/90787/1 "2024-09-08T08:50:39Z")

</div>

Hello all,

I'd like to create (or ideally use an existing one) a Node-RED node that:

1. Receives an input.
2. Publishes a message to either Kafka or MQTT.
3. Listens to a topic until it receives a response to the original message, processed by an external service.
4. Writes the response to the output.

Does such a node already exist? Is it possible to create in Node-RED?

```auto
# pseudo code: input -> my node -> output

# my node:
id=random()
pub("my_topic_input", id, input)
// other messages may arrive but for other id. They should not be lost as they belongs to other inputs
output = await sub("my_topic_output", id)
return output

```

---

<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:** [8 September 2024 09:00 UTC](https://discourse.nodered.org/t/creating-an-async-node-that-uses-pub-sub/90787/2 "2024-09-08T09:00:53Z")

</div>

None of this requires the development of a new node.

You can create a (reusable) subroutine (via a link-call node) or a subflow containing the necessary existing nodes that do what you are asking.

If your broker is v5 ready (most are) and your end device uses the `response topic` pattern, you can take advantage of that also.

---

<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:** [8 September 2024 09:10 UTC](https://discourse.nodered.org/t/creating-an-async-node-that-uses-pub-sub/90787/3 "2024-09-08T09:10:34Z")

</div>

Where MQTT is not in use (and for general logic)  
you can use a combination of `change` and `switch` nodes

- When you send the request
  - Add your ID (that is being sent) to a flow variable (via `change` node)

![Screenshot 2024-09-08 at 10.07.05](https://us1.discourse-cdn.com/flex026/uploads/nodered/original/3X/5/3/537d30d7add136bdd1e28151e7c20f221a355d2b.png)

Then,... when you're receiving responses back, only pass it through if it matches the recently set ID (via the `switch` node)

![Screenshot 2024-09-08 at 10.08.23](https://us1.discourse-cdn.com/flex026/uploads/nodered/original/3X/f/a/fa442c7eb49028b6c76d1313fb501f66d549ea1e.png)

and like @Steve-Mcl said - you can wrap this is in a unit for reuse

---

<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:** [8 September 2024 09:12 UTC](https://discourse.nodered.org/t/creating-an-async-node-that-uses-pub-sub/90787/4 "2024-09-08T09:12:52Z")

</div>

Be wary of using context in async things like this. Fast consecutive operations would have concurrency issues. It is doable via arrays or lookup that keep all outbound messages and perform a look up upon response etc. But the link-call mostly takes care of all that.

---

<div class="post-metadata">

**Author:** ![seb](https://sea2.discourse-cdn.com/flex026/user_avatar/discourse.nodered.org/seb/32/94157_2.png) [@seb](https://discourse.nodered.org/u/seb)\
**Post date:** [10 September 2024 11:25 UTC](https://discourse.nodered.org/t/creating-an-async-node-that-uses-pub-sub/90787/5 "2024-09-10T11:25:42Z")

</div>

thanks @Steve-Mcl ,

I've tried with link-call but without success... So the starting point was as simple as

 ![Screenshot 2024-09-10 at 12.22.48](https://us1.discourse-cdn.com/flex026/uploads/nodered/original/3X/2/7/27b5cae8f8cd7b6e7dced45561d0b30981885a95.png)

 ![Screenshot 2024-09-10 at 12.23.23](https://us1.discourse-cdn.com/flex026/uploads/nodered/original/3X/2/4/24cc90999c1b12fcf37908b109d70e76327535ae.png)

But I've received  
 ![Screenshot 2024-09-10 at 12.23.50](https://us1.discourse-cdn.com/flex026/uploads/nodered/original/3X/e/6/e64834f0201b5140fb6b26af7919e86a758f587d.png)

So my next temptative was like this:

 ![Screenshot 2024-09-10 at 12.24.40](https://us1.discourse-cdn.com/flex026/uploads/nodered/original/3X/a/9/a9d60a1e179e1b74831a95dfbbb4ac35b29641b6.png)

Trying to keep thosse node informations...

But I'm pretty sure I'm missing something... cannot be so complicated.

Just wondering if it woud be simpler just to code it in a node ?

---

<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:** [9 November 2024 11:26 UTC](https://discourse.nodered.org/t/creating-an-async-node-that-uses-pub-sub/90787/6 "2024-11-09T11:26:43Z")

</div>

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