Wildcard Subscriptions

There is no special code to subscribe with a wildcard subject. Wildcards are a normal part of the subject name. However, it is a common technique to use the subject provided with the incoming message to determine what to do with the message.

For example, you can subscribe using * and then act based on the actual subject.

{% tabs %} {% tab title=”Go” %}

  1. nc, err := nats.Connect("demo.nats.io")
  2. if err != nil {
  3.     log.Fatal(err)
  4. }
  5. defer nc.Close()
  7. // Use a WaitGroup to wait for 2 messages to arrive
  8. wg := sync.WaitGroup{}
  9. wg.Add(2)
  11. // Subscribe
  12. if _, err := nc.Subscribe("time.*.east", func(m *nats.Msg) {
  13.     log.Printf("%s: %s", m.Subject, m.Data)
  14.     wg.Done()
  15. }); err != nil {
  16.     log.Fatal(err)
  17. }
  19. // Wait for the 2 messages to come in
  20. wg.Wait()

{% endtab %}

{% tab title=”Java” %}

  1. Connection nc = Nats.connect("nats://demo.nats.io:4222");
  3. // Use a latch to wait for 2 messages to arrive
  4. CountDownLatch latch = new CountDownLatch(2);
  6. // Create a dispatcher and inline message handler
  7. Dispatcher d = nc.createDispatcher((msg) -> {
  8.     String subject = msg.getSubject();
  9.     String str = new String(msg.getData(), StandardCharsets.UTF_8);
  10.     System.out.println(subject + ": " + str);
  11.     latch.countDown();
  12. });
  14. // Subscribe
  15. d.subscribe("time.*.east");
  17. // Wait for messages to come in
  18. latch.await();
  20. // Close the connection
  21. nc.close();

{% endtab %}

{% tab title=”JavaScript” %}

  1. nc.subscribe("time.us.*", (_err, msg) => {
  2.     // converting timezones correctly in node requires a library
  3.     // this doesn't take into account *many* things.
  4.     let time;
  5.     switch (msg.subject) {
  6.       case "time.us.east":
  7.         time = new Date().toLocaleTimeString("en-us", {
  8.           timeZone: "America/New_York",
  9.         });
  10.         break;
  11.       case "time.us.central":
  12.         time = new Date().toLocaleTimeString("en-us", {
  13.           timeZone: "America/Chicago",
  14.         });
  15.         break;
  16.       case "time.us.mountain":
  17.         time = new Date().toLocaleTimeString("en-us", {
  18.           timeZone: "America/Denver",
  19.         });
  20.         break;
  21.       case "time.us.west":
  22.         time = new Date().toLocaleTimeString("en-us", {
  23.           timeZone: "America/Los_Angeles",
  24.         });
  25.         break;
  26.       default:
  27.         time = "I don't know what you are talking about Willis";
  28.     }
  29.     t.log(subject, time);
  30. });

{% endtab %}

{% tab title=”Python” %}

  1. nc = NATS()
  3. await nc.connect(servers=["nats://demo.nats.io:4222"])
  5. # Use queue to wait for 2 messages to arrive
  6. queue = asyncio.Queue()
  7. async def cb(msg):
  8.   await queue.put_nowait(msg)
  10. await nc.subscribe("time.*.east", cb=cb)
  12. # Send 2 messages and wait for them to come in
  13. await nc.publish("time.A.east", b'A')
  14. await nc.publish("time.B.east", b'B')
  16. msg_A = await queue.get()
  17. msg_B = await queue.get()
  19. print("Msg A:", msg_A)
  20. print("Msg B:", msg_B)

{% endtab %}

{% tab title=”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.*.east") do |msg, reply|
  9.       f.resume Time.now
  10.     end
  12.     nc.publish("time.A.east", "A")
  13.     nc.publish("time.B.east", "B")
  15.     # Use the response
  16.     msg_A = Fiber.yield
  17.     puts "Msg A: #{msg_A}"
  19.     msg_B = Fiber.yield
  20.     puts "Msg B: #{msg_B}"
  22.   end.resume
  23. end

{% endtab %}

{% tab title=”C” %}

  1. static void
  2. onMsg(natsConnection *conn, natsSubscription *sub, natsMsg *msg, void *closure)
  3. {
  4.     printf("Received msg: %s - %.*s\n",
  5.            natsMsg_GetSubject(msg),
  6.            natsMsg_GetDataLength(msg),
  7.            natsMsg_GetData(msg));
  9.     // Need to destroy the message!
  10.     natsMsg_Destroy(msg);
  11. }
  14. (...)
  16. natsConnection      *conn = NULL;
  17. natsSubscription    *sub  = NULL;
  18. natsStatus          s;
  20. s = natsConnection_ConnectTo(&conn, NATS_DEFAULT_URL);
  21. if (s == NATS_OK)
  22.     s = natsConnection_Subscribe(&sub, conn, "time.*.east", onMsg, NULL);
  24. (...)
  27. // Destroy objects that were created
  28. natsSubscription_Destroy(sub);
  29. natsConnection_Destroy(conn);

{% endtab %} {% endtabs %}

or do something similar with >:

{% tabs %} {% tab title=”Go” %}

  1. nc, err := nats.Connect("demo.nats.io")
  2. if err != nil {
  3.     log.Fatal(err)
  4. }
  5. defer nc.Close()
  7. // Use a WaitGroup to wait for 4 messages to arrive
  8. wg := sync.WaitGroup{}
  9. wg.Add(4)
  11. // Subscribe
  12. if _, err := nc.Subscribe("time.>", func(m *nats.Msg) {
  13.     log.Printf("%s: %s", m.Subject, m.Data)
  14.     wg.Done()
  15. }); err != nil {
  16.     log.Fatal(err)
  17. }
  19. // Wait for the 4 messages to come in
  20. wg.Wait()
  22. // Close the connection
  23. nc.Close()

{% endtab %}

{% tab title=”Java” %}

  1. Connection nc = Nats.connect("nats://demo.nats.io:4222");
  3. // Use a latch to wait for 4 messages to arrive
  4. CountDownLatch latch = new CountDownLatch(4);
  6. // Create a dispatcher and inline message handler
  7. Dispatcher d = nc.createDispatcher((msg) -> {
  8.     String subject = msg.getSubject();
  9.     String str = new String(msg.getData(), StandardCharsets.UTF_8);
  10.     System.out.println(subject + ": " + str);
  11.     latch.countDown();
  12. });
  14. // Subscribe
  15. d.subscribe("time.>");
  17. // Wait for messages to come in
  18. latch.await();
  20. // Close the connection
  21. nc.close();

{% endtab %}

{% tab title=”JavaScript” %}

  1. let nc = NATS.connect({
  2.     url: "nats://demo.nats.io:4222"
  3. });
  5. nc.subscribe('time.>', (msg, reply, subject) => {
  6.     // converting timezones correctly in node requires a library
  7.     // this doesn't take into account *many* things.
  8.     let time = "";
  9.     switch (subject) {
  10.         case 'time.us.east':
  11.             time = new Date().toLocaleTimeString("en-us", {timeZone: "America/New_York"});
  12.             break;
  13.         case 'time.us.central':
  14.             time = new Date().toLocaleTimeString("en-us", {timeZone: "America/Chicago"});
  15.             break;
  16.         case 'time.us.mountain':
  17.             time = new Date().toLocaleTimeString("en-us", {timeZone: "America/Denver"});
  18.             break;
  19.         case 'time.us.west':
  20.             time = new Date().toLocaleTimeString("en-us", {timeZone: "America/Los_Angeles"});
  21.             break;
  22.         default:
  23.             time = "I don't know what you are talking about Willis";
  24.     }
  25.     t.log(subject, time);
  26. });

{% endtab %}

{% tab title=”Python” %}

  1. nc = NATS()
  3. await nc.connect(servers=["nats://demo.nats.io:4222"])
  5. # Use queue to wait for 4 messages to arrive
  6. queue = asyncio.Queue()
  7. async def cb(msg):
  8.   await queue.put(msg)
  10. await nc.subscribe("time.>", cb=cb)
  12. # Send 2 messages and wait for them to come in
  13. await nc.publish("time.A.east", b'A')
  14. await nc.publish("time.B.east", b'B')
  15. await nc.publish("time.C.west", b'C')
  16. await nc.publish("time.D.west", b'D')
  18. for i in range(0, 4):
  19.   msg = await queue.get()
  20.   print("Msg:", msg)
  22. await nc.close()

{% endtab %}

{% tab title=”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.to_f
  10.     end
  12.     nc.publish("time.A.east", "A")
  13.     nc.publish("time.B.east", "B")
  14.     nc.publish("time.C.west", "C")
  15.     nc.publish("time.D.west", "D")
  17.     # Use the response
  18.     4.times do 
  19.       msg = Fiber.yield
  20.       puts "Msg: #{msg}"
  21.     end
  22.   end.resume
  23. end

{% endtab %}

{% tab title=”TypeScript” %}

  1. await nc.subscribe('time.>', (err, msg) => {
  2.     // converting timezones correctly in node requires a library
  3.     // this doesn't take into account *many* things.
  4.     let time = "";
  5.     switch (msg.subject) {
  6.         case 'time.us.east':
  7.             time = new Date().toLocaleTimeString("en-us", {timeZone: "America/New_York"});
  8.             break;
  9.         case 'time.us.central':
  10.             time = new Date().toLocaleTimeString("en-us", {timeZone: "America/Chicago"});
  11.             break;
  12.         case 'time.us.mountain':
  13.             time = new Date().toLocaleTimeString("en-us", {timeZone: "America/Denver"});
  14.             break;
  15.         case 'time.us.west':
  16.             time = new Date().toLocaleTimeString("en-us", {timeZone: "America/Los_Angeles"});
  17.             break;
  18.         default:
  19.             time = "I don't know what you are talking about Willis";
  20.     }
  21.     t.log(msg.subject, time);
  22. });

{% endtab %}

{% tab title=”C” %}

  1. static void
  2. onMsg(natsConnection *conn, natsSubscription *sub, natsMsg *msg, void *closure)
  3. {
  4.     printf("Received msg: %s - %.*s\n",
  5.            natsMsg_GetSubject(msg),
  6.            natsMsg_GetDataLength(msg),
  7.            natsMsg_GetData(msg));
  9.     // Need to destroy the message!
  10.     natsMsg_Destroy(msg);
  11. }
  14. (...)
  16. natsConnection      *conn = NULL;
  17. natsSubscription    *sub  = NULL;
  18. natsStatus          s;
  20. s = natsConnection_ConnectTo(&conn, NATS_DEFAULT_URL);
  21. if (s == NATS_OK)
  22.     s = natsConnection_Subscribe(&sub, conn, "time.>", onMsg, NULL);
  24. (...)
  27. // Destroy objects that were created
  28. natsSubscription_Destroy(sub);
  29. natsConnection_Destroy(conn);

{% endtab %} {% endtabs %}

The following example can be used to test these two subscribers. The * subscriber should receive at most 2 messages, while the > subscriber receives 4. More importantly the time.*.east subscriber won’t receive on time.us.east.atlanta because that won’t match.

{% tabs %} {% tab title=”Go” %}

  1. nc, err := nats.Connect("demo.nats.io")
  2. if err != nil {
  3.     log.Fatal(err)
  4. }
  5. defer nc.Close()
  7. zoneID, err := time.LoadLocation("America/New_York")
  8. if err != nil {
  9.     log.Fatal(err)
  10. }
  11. now := time.Now()
  12. zoneDateTime := now.In(zoneID)
  13. formatted := zoneDateTime.String()
  15. nc.Publish("time.us.east", []byte(formatted))
  16. nc.Publish("time.us.east.atlanta", []byte(formatted))
  18. zoneID, err = time.LoadLocation("Europe/Warsaw")
  19. if err != nil {
  20.     log.Fatal(err)
  21. }
  22. zoneDateTime = now.In(zoneID)
  23. formatted = zoneDateTime.String()
  25. nc.Publish("time.eu.east", []byte(formatted))
  26. nc.Publish("time.eu.east.warsaw", []byte(formatted))

{% endtab %}

{% tab title=”Java” %}

  1. Connection nc = Nats.connect("nats://demo.nats.io:4222");
  2. ZoneId zoneId = ZoneId.of("America/New_York");
  3. ZonedDateTime zonedDateTime = ZonedDateTime.ofInstant(Instant.now(), zoneId);
  4. String formatted = zonedDateTime.format(DateTimeFormatter.ISO_ZONED_DATE_TIME);
  6. nc.publish("time.us.east", formatted.getBytes(StandardCharsets.UTF_8));
  7. nc.publish("time.us.east.atlanta", formatted.getBytes(StandardCharsets.UTF_8));
  9. zoneId = ZoneId.of("Europe/Warsaw");
  10. zonedDateTime = ZonedDateTime.ofInstant(Instant.now(), zoneId);
  11. formatted = zonedDateTime.format(DateTimeFormatter.ISO_ZONED_DATE_TIME);
  12. nc.publish("time.eu.east", formatted.getBytes(StandardCharsets.UTF_8));
  13. nc.publish("time.eu.east.warsaw", formatted.getBytes(StandardCharsets.UTF_8));
  15. nc.flush(Duration.ZERO);
  16. nc.close();

{% endtab %}

{% tab title=”JavaScript” %}

  1. nc.publish('time.us.east');
  2. nc.publish('time.us.central');
  3. nc.publish('time.us.mountain');
  4. nc.publish('time.us.west');

{% endtab %}

{% tab title=”Python” %}

  1. nc = NATS()
  3. await nc.connect(servers=["nats://demo.nats.io:4222"])
  5. await nc.publish("time.us.east", b'...')
  6. await nc.publish("time.us.east.atlanta", b'...')
  8. await nc.publish("time.eu.east", b'...')
  9. await nc.publish("time.eu.east.warsaw", b'...')
  11. await nc.close()

{% endtab %}

{% tab title=”Ruby” %}

  1. NATS.start do |nc|
  2.    nc.publish("time.us.east", '...')
  3.    nc.publish("time.us.east.atlanta", '...')
  5.    nc.publish("time.eu.east", '...')
  6.    nc.publish("time.eu.east.warsaw", '...')
  8.    nc.drain
  9. end

{% endtab %}

{% tab title=”TypeScript” %}

  1. nc.publish('time.us.east');
  2. nc.publish('time.us.central');
  3. nc.publish('time.us.mountain');
  4. nc.publish('time.us.west');

{% endtab %} {% endtabs %}