Replying to a Message

Incoming messages have an optional reply-to field. If that field is set, it will contain a subject to which a reply is expected.

For example, the following code will listen for that request and respond with the time.

Go

  1. nc, err := nats.Connect("demo.nats.io")
  2. if err != nil {
  3.     log.Fatal(err)
  4. }
  5. defer nc.Close()
  7. // Subscribe
  8. sub, err := nc.SubscribeSync("time")
  9. if err != nil {
  10.     log.Fatal(err)
  11. }
  13. // Read a message
  14. msg, err := sub.NextMsg(10 * time.Second)
  15. if err != nil {
  16.     log.Fatal(err)
  17. }
  19. // Get the time
  20. timeAsBytes := []byte(time.Now().String())
  22. // Send the time as the response.
  23. msg.Respond(timeAsBytes)

Java

  1. Connection nc = Nats.connect("nats://demo.nats.io:4222");
  3. // Subscribe
  4. Subscription sub = nc.subscribe("time");
  6. // Read a message
  7. Message msg = sub.nextMessage(Duration.ZERO);
  9. // Get the time
  10. Calendar cal = Calendar.getInstance();
  11. SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss");
  12. byte[] timeAsBytes = sdf.format(cal.getTime()).getBytes(StandardCharsets.UTF_8);
  14. // Send the time
  15. nc.publish(msg.getReplyTo(), timeAsBytes);
  17. // Flush and close the connection
  18. nc.flush(Duration.ZERO);
  19. nc.close();

JavaScript

  1. let nc = NATS.connect({
  2.     url: "nats://demo.nats.io:4222"
  3. });
  5. // set up a subscription to process a request
  6. nc.subscribe('time', (msg, reply) => {
  7.     if (msg.reply) {
  8.         nc.publish(msg.reply, new Date().toLocaleTimeString());
  9.     }
  10. });

Python

  1. nc = NATS()
  3. await nc.connect(servers=["nats://demo.nats.io:4222"])
  5. future = asyncio.Future()
  7. async def cb(msg):
  8.   nonlocal future
  9.   future.set_result(msg)
  11. await nc.subscribe("time", cb=cb)
  13. await nc.publish_request("time", new_inbox(), b'What is the time?')
  14. await nc.flush()
  16. # Read the message
  17. msg = await asyncio.wait_for(future, 1)
  19. # Send the time
  20. time_as_bytes = "{}".format(datetime.now()).encode()
  21. await nc.publish(msg.reply, time_as_bytes)

Ruby

  1. require 'nats/client'
  2. require 'fiber'
  4. NATS.start(servers:["nats://127.0.0.1:4222"]) do |nc|
  5.   Fiber.new do
  6.     f = Fiber.current
  8.     nc.subscribe("time") do |msg, reply|
  9.       f.resume Time.now
  10.     end
  12.     nc.publish("time", 'What is the time?', NATS.create_inbox)
  14.     # Use the response
  15.     msg = Fiber.yield
  16.     puts "Reply: #{msg}"
  18.   end.resume
  19. end

TypeScript

  1. // set up a subscription to process a request
  2. await nc.subscribe('time', (err, msg) => {
  3.     if (msg.reply) {
  4.         nc.publish(msg.reply, new Date().toLocaleTimeString());
  5.     } else {
  6.         t.log('got a request for the time, but no reply subject was set.');
  7.     }
  8. });

C

  1. natsConnection      *conn      = NULL;
  2. natsSubscription    *sub       = NULL;
  3. natsMsg             *msg       = NULL;
  4. natsStatus          s          = NATS_OK;
  6. s = natsConnection_ConnectTo(&conn, NATS_DEFAULT_URL);
  8. // Subscribe
  9. if (s == NATS_OK)
  10.     s = natsConnection_SubscribeSync(&sub, conn, "time");
  12. // Wait for messages
  13. if (s == NATS_OK)
  14.     s = natsSubscription_NextMsg(&msg, sub, 10000);
  16. if (s == NATS_OK)
  17. {
  18.     char buf[64];
  20.     snprintf(buf, sizeof(buf), "%lld", nats_Now());
  22.     // Send the time as a response
  23.     s = natsConnection_Publish(conn, natsMsg_GetReply(msg), buf, (int) strlen(buf));
  25.     // Destroy message that was received
  26.     natsMsg_Destroy(msg);
  27. }
  29. (...)
  31. // Destroy objects that were created
  32. natsSubscription_Destroy(sub);
  33. natsConnection_Destroy(conn);