Wildcard Subscriptions

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

    Go

    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();

    JavaScript

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

    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)

    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

    TypeScript

    1. await 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", {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.             break;
    18.             time = "I don't know what you are talking about Willis";
    19.     }
    20.     console.log(msg.subject, time);
    21. });

    or do something similar with >:

    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()

    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();

    JavaScript

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

    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()

    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.       f.resume Time.now.to_f
    9.     end
    11.     nc.publish("time.A.east", "A")
    12.     nc.publish("time.B.east", "B")
    13.     nc.publish("time.C.west", "C")
    14.     nc.publish("time.D.west", "D")
    16.     # Use the response
    17.     4.times do 
    18.       msg = Fiber.yield
    19.       puts "Msg: #{msg}"
    20.     end
    21.   end.resume
    22. end

    TypeScript

    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);

    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.

    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))

    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();

    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');

    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()

    Ruby

    TypeScript

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