JetStream Consumers

Consumers are how client applications get the messages stored in the streams. You can have many consumers on a single stream. Consumers are like a view on a stream, can filter messages and have some state (maintained by the servers) associated with them.

Consumers can be ‘durable’ or ‘ephemeral’.

Ephemeral consumers

Ephemeral consumers are meant to be used by a single instance of an application (e.g. to get its own replay of the messages in the stream).

Ephemeral consumers are not meant to last ‘forever’, they are defined automatically at subscription time by the client library and disappear after the application disconnect.

You (automatically) create an ephemeral consumer when you call the js.Subscribe function without specifying the Durable or Bind subscription options. Calling Drain on that subscription automatically deletes the underlying ephemeral consumer. You can also explicitly create an ephemeral consumer by not passing a durable name option to the jsm.AddConsumer call.

Ephemeral consumers otherwise have the same control over message acknowledged and re-delivery as durable consumers.

Ordered Consumers

The example below uses an ordered consumer, a convenient default type of push consumers designed for applications that want to efficiently consume a stream for data inspection or analysis.

  • Always ephemeral
  • Auto acknowledgment (no re-delivery)
  • Single threaded dispatching

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

  1. func ExampleJetStream() {
  2.     nc, err := nats.Connect("localhost")
  3.     if err != nil {
  4.         log.Fatal(err)
  5.     }
  7.     // Use the JetStream context to produce and consumer messages
  8.     // that have been persisted.
  9.     js, err := nc.JetStream(nats.PublishAsyncMaxPending(256))
  10.     if err != nil {
  11.         log.Fatal(err)
  12.     }
  14.     js.AddStream(&nats.StreamConfig{
  15.         Name:     "FOO",
  16.         Subjects: []string{"foo"},
  17.     })
  19.     js.Publish("foo", []byte("Hello JS!"))
  21.     // ordered push consumer
  22.     js.Subscribe("foo", func(msg *nats.Msg) {
  23.         meta, _ := msg.Metadata()
  24.         fmt.Printf("Stream Sequence  : %v\n", meta.Sequence.Stream)
  25.         fmt.Printf("Consumer Sequence: %v\n", meta.Sequence.Consumer)
  26.     }, nats.OrderedConsumer())
  27. }

{% endtab %} {% tab title=”Java” %}

  1. package io.nats.examples.jetstream;
  3. import io.nats.client.*;
  4. import io.nats.client.api.PublishAck;
  5. import io.nats.client.impl.NatsMessage;
  6. import io.nats.examples.ExampleArgs;
  7. import io.nats.examples.ExampleUtils;
  9. import java.nio.charset.StandardCharsets;
  10. import java.time.Duration;
  11. import java.time.temporal.TemporalUnit;
  13. public class myExample {
  14.  public static void main(String[] args) {
  15.   final String subject = "foo";
  17.   try (Connection nc = Nats.connect(ExampleUtils.createExampleOptions("localhost"))) {
  19.    // Create a JetStream context.  This hangs off the original connection
  20.    // allowing us to produce data to streams and consume data from
  21.    // JetStream consumers.
  22.    JetStream js = nc.jetStream();
  24.    // This example assumes there is a stream already created on subject "foo" and some messages already stored in that stream
  26.    // create our message handler.
  27.    MessageHandler handler = msg -> {
  29.     System.out.println("\nMessage Received:");
  31.     if (msg.hasHeaders()) {
  32.      System.out.println("  Headers:");
  33.      for (String key : msg.getHeaders().keySet()) {
  34.       for (String value : msg.getHeaders().get(key)) {
  35.        System.out.printf("    %s: %s\n", key, value);
  36.       }
  37.      }
  38.     }
  40.     System.out.printf("  Subject: %s\n  Data: %s\n",
  41.             msg.getSubject(), new String(msg.getData(), StandardCharsets.UTF_8));
  42.     System.out.println("  " + msg.metaData());
  43.    };
  45.    Dispatcher dispatcher = nc.createDispatcher();
  46.    PushSubscribeOptions pso = PushSubscribeOptions.builder().ordered(true).build();
  47.    JetStreamSubscription sub = js.subscribe(subject, dispatcher, handler, false, pso);
  49.    Thread.sleep(100);
  51.    sub.drain(Duration.ofMillis(100));
  53.    nc.drain(Duration.ofMillis(100));
  54.   }
  55.   catch(Exception e)
  56.   {
  57.    e.printStackTrace();
  58.   }
  59.  }
  60. }

{% endtab %} {% tab title=”JavaScript” %}

  1. import { connect, consumerOpts } from "../../src/mod.ts";
  3. const nc = await connect();
  4. const js = nc.jetstream();
  6. // note the consumer is not a durable - so when after the
  7. // subscription ends, the server will auto destroy the
  8. // consumer
  9. const opts = consumerOpts();
  10. opts.manualAck();
  11. opts.maxMessages(2);
  12. opts.deliverTo("xxx");
  13. const sub = await js.subscribe("a.>", opts);
  14. await (async () => {
  15.   for await (const m of sub) {
  16.     console.log(m.seq, m.subject);
  17.     m.ack();
  18.   }
  19. })();
  21. await nc.close();

{% endtab %} {% tab title= “Python” %}

  1. import asyncio
  3. import nats
  4. from nats.errors import TimeoutError
  7. async def main():
  8.     nc = await nats.connect("localhost")
  10.     # Create JetStream context.
  11.     js = nc.jetstream()
  13.     # Create ordered consumer with flow control and heartbeats
  14.     # that auto resumes on failures.
  15.     osub = await js.subscribe("foo", ordered_consumer=True)
  16.     data = bytearray()
  18.     while True:
  19.         try:
  20.             msg = await osub.next_msg()
  21.             data.extend(msg.data)
  22.         except TimeoutError:
  23.             break
  24.     print("All data in stream:", len(data))
  26.     await nc.close()
  28. if __name__ == '__main__':
  29.     asyncio.run(main())

{% endtab %} {% endtabs %}

Durable consumers

Durable consumers are meant to be used by multiple instances of an application, either to distribute and scale out the processing, or to persist the position of the consumer over the stream between runs of an application.

Durable consumers as the name implies are meant to last ‘forever’ and are typically created and deleted administratively rather than by the application code which only needs to specify the durable’s well known name to use it.

You create a durable consumer using the nats consumer add CLI tool command, or programmatically by passing a durable name option to the subscription creation call.

Push and Pull consumers

Clients implement two implementations of consumers identified as ‘push’ or ‘pull’. Push consumers receive messages on a specific subject where message flow is controlled by the server. Load balancing is supported through NATS core queue groups. The messages from the stream are distributed automatically between the subscribing clients to the push consumers.

Pull consumers request messages explicitly from the server in batches, giving the client full control over dispatching, flow control, pending (unacknowledged) messages and load balancing. Pull consuming client make fetch() calls in a dispatch loop.

{% hint style=”info” %}We recommend pull consumers for new projects. In particular when scalability, detailed flow control or error handling are a concern. {% endhint %}

Pull consumers create less CPU load on the NATS servers and therefore scale better (note that the push consumers are still quite fast and scalable, you may only notice the difference between the two if you have sustained high message rates).

Pull

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

  1. func ExampleJetStream() {
  2.     nc, err := nats.Connect("localhost")
  3.     if err != nil {
  4.         log.Fatal(err)
  5.     }
  7.     // Use the JetStream context to produce and consumer messages
  8.     // that have been persisted.
  9.     js, err := nc.JetStream(nats.PublishAsyncMaxPending(256))
  10.     if err != nil {
  11.         log.Fatal(err)
  12.     }
  14.     js.AddStream(&nats.StreamConfig{
  15.         Name:     "FOO",
  16.         Subjects: []string{"foo"},
  17.     })
  19.     js.Publish("foo", []byte("Hello JS!"))
  21.     // Publish messages asynchronously.
  22.     for i := 0; i < 500; i++ {
  23.         js.PublishAsync("foo", []byte("Hello JS Async!"))
  24.     }
  25.     select {
  26.     case <-js.PublishAsyncComplete():
  27.     case <-time.After(5 * time.Second):
  28.         fmt.Println("Did not resolve in time")
  29.     }
  31.     // Create Pull based consumer with maximum 128 inflight.
  32.     sub, _ := js.PullSubscribe("foo", "wq", nats.PullMaxWaiting(128))
  34.     ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
  35.     defer cancel()
  37.     for {
  38.         select {
  39.         case <-ctx.Done():
  40.             return
  41.         default:
  42.         }
  44.         // Fetch will return as soon as any message is available rather than wait until the full batch size is available, using a batch size of more than 1 allows for higher throughput when needed.
  45.         msgs, _ := sub.Fetch(10, nats.Context(ctx))
  46.         for _, msg := range msgs {
  47.             msg.Ack()
  48.         }
  49.     }
  50. }

{% endtab %} {% tab title=”Java” %}

  1. package io.nats.examples.jetstream;
  3. import io.nats.client.*;
  4. import io.nats.examples.ExampleArgs;
  5. import io.nats.examples.ExampleUtils;
  7. import java.time.Duration;
  9. import static io.nats.examples.jetstream.NatsJsUtils.createStreamExitWhenExists;
  10. import static io.nats.examples.jetstream.NatsJsUtils.publishInBackground;
  12. /**
  13.  * This example will demonstrate basic use of a pull subscription of:
  14.  * batch size only pull: <code>pull(int batchSize)</code>
  15.  */
  16. public class NatsJsPullSubBatchSize {
  17.     static final String usageString =
  18.         "\nUsage: java -cp <classpath> NatsJsPullSubBatchSize [-s server] [-strm stream] [-sub subject] [-dur durable] [-mcnt msgCount]"
  19.             + "\n\nDefault Values:"
  20.             + "\n   [-strm] pull-stream"
  21.             + "\n   [-sub]  pull-subject"
  22.             + "\n   [-dur]  pull-durable"
  23.             + "\n   [-mcnt] 20"
  24.             + "\n\nUse tls:// or opentls:// to require tls, via the Default SSLContext\n"
  25.             + "\nSet the environment variable NATS_NKEY to use challenge response authentication by setting a file containing your private key.\n"
  26.             + "\nSet the environment variable NATS_CREDS to use JWT/NKey authentication by setting a file containing your user creds.\n"
  27.             + "\nUse the URL in the -s server parameter for user/pass/token authentication.\n";
  29.     public static void main(String[] args) {
  30.         ExampleArgs exArgs = ExampleArgs.builder("Pull Subscription using primitive Batch Size", args, usageString)
  31.                 .defaultStream("pull-stream")
  32.                 .defaultSubject("pull-subject")
  33.                 .defaultDurable("pull-durable")
  34.                 .defaultMsgCount(15)
  35.                 .build();
  37.         try (Connection nc = Nats.connect(ExampleUtils.createExampleOptions(exArgs.server))) {
  38.             // Create a JetStreamManagement context.
  39.             JetStreamManagement jsm = nc.jetStreamManagement();
  41.             // Use the utility to create a stream stored in memory.
  42.             createStreamExitWhenExists(jsm, exArgs.stream, exArgs.subject);
  44.             // Create our JetStream context.
  45.             JetStream js = nc.jetStream();
  47.             // start publishing the messages, don't wait for them to finish, simulating an outside producer
  48.             publishInBackground(js, exArgs.subject, "pull-message", exArgs.msgCount);
  50.             // Build our subscription options. Durable is REQUIRED for pull based subscriptions
  51.             PullSubscribeOptions pullOptions = PullSubscribeOptions.builder()
  52.                 .durable(exArgs.durable) // required
  53.                 .build();
  55.             // subscribe
  56.             JetStreamSubscription sub = js.subscribe(exArgs.subject, pullOptions);
  57.             nc.flush(Duration.ofSeconds(1));
  59.             int red = 0;
  60.             while (red < exArgs.msgCount) {
  61.                 sub.pull(10);
  62.                 Message m = sub.nextMessage(Duration.ofSeconds(1)); // first message
  63.                 while (m != null) {
  64.                     if (m.isJetStream()) {
  65.                         red++; // process message
  66.                         System.out.println("" + red + ". " + m);
  67.                         m.ack();
  68.                     }
  69.                     m = sub.nextMessage(Duration.ofMillis(100)); // other messages should already be on the client
  70.                 }
  71.             }
  73.             // delete the stream since we are done with it.
  74.             jsm.deleteStream(exArgs.stream);
  75.         }
  76.         catch (Exception e) {
  77.             e.printStackTrace();
  78.         }
  79.     }
  80. }

{% endtab %} {% tab title=”JavaScript” %}

  1. import { AckPolicy, connect, nanos } from "../../src/mod.ts";
  2. import { nuid } from "../../nats-base-client/nuid.ts";
  4. const nc = await connect();
  6. const stream = nuid.next();
  7. const subj = nuid.next();
  8. const durable = nuid.next();
  10. const jsm = await nc.jetstreamManager();
  11. await jsm.streams.add({ name: stream, subjects: [subj] });
  13. const js = nc.jetstream();
  14. await js.publish(subj);
  15. await js.publish(subj);
  16. await js.publish(subj);
  17. await js.publish(subj);
  19. const psub = await js.pullSubscribe(subj, {
  20.   mack: true,
  21.   // artificially low ack_wait, to show some messages
  22.   // not getting acked being redelivered
  23.   config: {
  24.     durable_name: durable,
  25.     ack_policy: AckPolicy.Explicit,
  26.     ack_wait: nanos(4000),
  27.   },
  28. });
  30. (async () => {
  31.   for await (const m of psub) {
  32.     console.log(
  33.       `[${m.seq}] ${
  34.         m.redelivered ? `- redelivery ${m.info.redeliveryCount}` : ""
  35.       }`
  36.     );
  37.     if (m.seq % 2 === 0) {
  38.       m.ack();
  39.     }
  40.   }
  41. })();
  43. const fn = () => {
  44.   console.log("[PULL]");
  45.   psub.pull({ batch: 1000, expires: 10000 });
  46. };
  48. // do the initial pull
  49. fn();
  50. // and now schedule a pull every so often
  51. const interval = setInterval(fn, 10000); // and repeat every 2s
  53. setTimeout(() => {
  54.   clearInterval(interval);
  55.   nc.drain();
  56. }, 20000);

{% endtab %} {% tab title= “Python” %}

  1. import asyncio
  3. import nats
  4. from nats.errors import TimeoutError
  6. async def main():
  7.     nc = await nats.connect("localhost")
  9.     # Create JetStream context.
  10.     js = nc.jetstream()
  12.     # Persist messages on 'foo's subject.
  13.     await js.add_stream(name="sample-stream", subjects=["foo"])
  15.     for i in range(0, 10):
  16.         ack = await js.publish("foo", f"hello world: {i}".encode())
  17.         print(ack)
  19.     # Create pull based consumer on 'foo'.
  20.     psub = await js.pull_subscribe("foo", "psub")
  22.     # Fetch and ack messagess from consumer.
  23.     for i in range(0, 10):
  24.         msgs = await psub.fetch(1)
  25.         for msg in msgs:
  26.             print(msg)
  28.     await nc.close()
  30. if __name__ == '__main__':
  31.     asyncio.run(main())

{% endtab %} {% tab title=”C” %}

  1. #include "examples.h"
  3. static const char *usage = ""\
  4. "-gd            use global message delivery thread pool\n" \
  5. "-sync          receive synchronously (default is asynchronous)\n" \
  6. "-pull          use pull subscription\n" \
  7. "-fc            enable flow control\n" \
  8. "-count         number of expected messages\n";
  10. static void
  11. onMsg(natsConnection *nc, natsSubscription *sub, natsMsg *msg, void *closure)
  12. {
  13.     if (print)
  14.         printf("Received msg: %s - %.*s\n",
  15.                natsMsg_GetSubject(msg),
  16.                natsMsg_GetDataLength(msg),
  17.                natsMsg_GetData(msg));
  19.     if (start == 0)
  20.         start = nats_Now();
  22.     // We should be using a mutex to protect those variables since
  23.     // they are used from the subscription's delivery and the main
  24.     // threads. For demo purposes, this is fine.
  25.     if (++count == total)
  26.         elapsed = nats_Now() - start;
  28.     // Since this is auto-ack callback, we don't need to ack here.
  29.     natsMsg_Destroy(msg);
  30. }
  32. static void
  33. asyncCb(natsConnection *nc, natsSubscription *sub, natsStatus err, void *closure)
  34. {
  35.     printf("Async error: %u - %s\n", err, natsStatus_GetText(err));
  37.     natsSubscription_GetDropped(sub, (int64_t*) &dropped);
  38. }
  40. int main(int argc, char **argv)
  41. {
  42.     natsConnection      *conn  = NULL;
  43.     natsStatistics      *stats = NULL;
  44.     natsOptions         *opts  = NULL;
  45.     natsSubscription    *sub   = NULL;
  46.     natsMsg             *msg   = NULL;
  47.     jsCtx               *js    = NULL;
  48.     jsErrCode           jerr   = 0;
  49.     jsOptions           jsOpts;
  50.     jsSubOptions        so;
  51.     natsStatus          s;
  52.     bool                delStream = false;
  54.     opts = parseArgs(argc, argv, usage);
  56.     printf("Created %s subscription on '%s'.\n",
  57.         (pull ? "pull" : (async ? "asynchronous" : "synchronous")), subj);
  59.     s = natsOptions_SetErrorHandler(opts, asyncCb, NULL);
  61.     if (s == NATS_OK)
  62.         s = natsConnection_Connect(&conn, opts);
  64.     if (s == NATS_OK)
  65.         s = jsOptions_Init(&jsOpts);
  67.     if (s == NATS_OK)
  68.         s = jsSubOptions_Init(&so);
  69.     if (s == NATS_OK)
  70.     {
  71.         so.Stream = stream;
  72.         so.Consumer = durable;
  73.         if (flowctrl)
  74.         {
  75.             so.Config.FlowControl = true;
  76.             so.Config.Heartbeat = (int64_t)1E9;
  77.         }
  78.     }
  80.     if (s == NATS_OK)
  81.         s = natsConnection_JetStream(&js, conn, &jsOpts);
  83.     if (s == NATS_OK)
  84.     {
  85.         jsStreamInfo    *si = NULL;
  87.         // First check if the stream already exists.
  88.         s = js_GetStreamInfo(&si, js, stream, NULL, &jerr);
  89.         if (s == NATS_NOT_FOUND)
  90.         {
  91.             jsStreamConfig  cfg;
  93.             // Since we are the one creating this stream, we can delete at the end.
  94.             delStream = true;
  96.             // Initialize the configuration structure.
  97.             jsStreamConfig_Init(&cfg);
  98.             cfg.Name = stream;
  99.             // Set the subject
  100.             cfg.Subjects = (const char*[1]){subj};
  101.             cfg.SubjectsLen = 1;
  102.             // Make it a memory stream.
  103.             cfg.Storage = js_MemoryStorage;
  104.             // Add the stream,
  105.             s = js_AddStream(&si, js, &cfg, NULL, &jerr);
  106.         }
  107.         if (s == NATS_OK)
  108.         {
  109.             printf("Stream %s has %" PRIu64 " messages (%" PRIu64 " bytes)\n",
  110.                 si->Config->Name, si->State.Msgs, si->State.Bytes);
  112.             // Need to destroy the returned stream object.
  113.             jsStreamInfo_Destroy(si);
  114.         }
  115.     }
  117.     if (s == NATS_OK)
  118.     {
  119.         if (pull)
  120.             s = js_PullSubscribe(&sub, js, subj, durable, &jsOpts, &so, &jerr);
  121.         else if (async)
  122.             s = js_Subscribe(&sub, js, subj, onMsg, NULL, &jsOpts, &so, &jerr);
  123.         else
  124.             s = js_SubscribeSync(&sub, js, subj, &jsOpts, &so, &jerr);
  125.     }
  126.     if (s == NATS_OK)
  127.         s = natsSubscription_SetPendingLimits(sub, -1, -1);
  129.     if (s == NATS_OK)
  130.         s = natsStatistics_Create(&stats);
  132.     if ((s == NATS_OK) && pull)
  133.     {
  134.         natsMsgList list;
  135.         int         i;
  137.         for (count = 0; (s == NATS_OK) && (count < total); )
  138.         {
  139.             s = natsSubscription_Fetch(&list, sub, 1024, 5000, &jerr);
  140.             if (s != NATS_OK)
  141.                 break;
  143.             if (start == 0)
  144.                 start = nats_Now();
  146.             count += (int64_t) list.Count;
  147.             for (i=0; (s == NATS_OK) && (i<list.Count); i++)
  148.                 s = natsMsg_Ack(list.Msgs[i], &jsOpts);
  150.             natsMsgList_Destroy(&list);
  151.         }
  152.     }
  153.     else if ((s == NATS_OK) && async)
  154.     {
  155.         while (s == NATS_OK)
  156.         {
  157.             if (count + dropped == total)
  158.                 break;
  160.             nats_Sleep(1000);
  161.         }
  162.     }
  163.     else if (s == NATS_OK)
  164.     {
  165.         for (count = 0; (s == NATS_OK) && (count < total); count++)
  166.         {
  167.             s = natsSubscription_NextMsg(&msg, sub, 5000);
  168.             if (s != NATS_OK)
  169.                 break;
  171.             if (start == 0)
  172.                 start = nats_Now();
  174.             s = natsMsg_Ack(msg, &jsOpts);
  175.             natsMsg_Destroy(msg);
  176.         }
  177.     }
  179.     if (s == NATS_OK)
  180.     {
  181.         printStats(STATS_IN|STATS_COUNT, conn, sub, stats);
  182.         printPerf("Received");
  183.     }
  184.     if (s == NATS_OK)
  185.     {
  186.         jsStreamInfo *si = NULL;
  188.         // Let's report some stats after the run
  189.         s = js_GetStreamInfo(&si, js, stream, NULL, &jerr);
  190.         if (s == NATS_OK)
  191.         {
  192.             printf("\nStream %s has %" PRIu64 " messages (%" PRIu64 " bytes)\n",
  193.                 si->Config->Name, si->State.Msgs, si->State.Bytes);
  195.             jsStreamInfo_Destroy(si);
  196.         }
  197.         if (delStream)
  198.         {
  199.             printf("\nDeleting stream %s: ", stream);
  200.             s = js_DeleteStream(js, stream, NULL, &jerr);
  201.             if (s == NATS_OK)
  202.                 printf("OK!");
  203.             printf("\n");
  204.         }
  205.     }
  206.     else
  207.     {
  208.         printf("Error: %u - %s - jerr=%u\n", s, natsStatus_GetText(s), jerr);
  209.         nats_PrintLastErrorStack(stderr);
  210.     }
  212.     // Destroy all our objects to avoid report of memory leak
  213.     jsCtx_Destroy(js);
  214.     natsStatistics_Destroy(stats);
  215.     natsSubscription_Destroy(sub);
  216.     natsConnection_Destroy(conn);
  217.     natsOptions_Destroy(opts);
  219.     // To silence reports of memory still in used with valgrind
  220.     nats_Close();
  222.     return 0;
  223. }

{% endtab %} {% endtabs %}

A push consumer can also be used in some other use cases such as without a queue group, or with no acknowledgement or cumulative acknowledgements.

Push

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

  1. func ExampleJetStream() {
  2.     nc, err := nats.Connect("localhost")
  3.     if err != nil {
  4.         log.Fatal(err)
  5.     }
  7.     // Use the JetStream context to produce and consumer messages
  8.     // that have been persisted.
  9.     js, err := nc.JetStream(nats.PublishAsyncMaxPending(256))
  10.     if err != nil {
  11.         log.Fatal(err)
  12.     }
  14.     js.AddStream(&nats.StreamConfig{
  15.         Name:     "FOO",
  16.         Subjects: []string{"foo"},
  17.     })
  19.     js.Publish("foo", []byte("Hello JS!"))
  21.     // Publish messages asynchronously.
  22.     for i := 0; i < 500; i++ {
  23.         js.PublishAsync("foo", []byte("Hello JS Async!"))
  24.     }
  25.     select {
  26.     case <-js.PublishAsyncComplete():
  27.     case <-time.After(5 * time.Second):
  28.         fmt.Println("Did not resolve in time")
  29.     }
  31.     // Create async consumer on subject 'foo'. Async subscribers
  32.     // ack a message once exiting the callback.
  33.     js.Subscribe("foo", func(msg *nats.Msg) {
  34.         meta, _ := msg.Metadata()
  35.         fmt.Printf("Stream Sequence  : %v\n", meta.Sequence.Stream)
  36.         fmt.Printf("Consumer Sequence: %v\n", meta.Sequence.Consumer)
  37.     })
  39.     // Async subscriber with manual acks.
  40.     js.Subscribe("foo", func(msg *nats.Msg) {
  41.         msg.Ack()
  42.     }, nats.ManualAck())
  44.     // Async queue subscription where members load balance the
  45.     // received messages together.
  46.     // If no consumer name is specified, either with nats.Bind()
  47.     // or nats.Durable() options, the queue name is used as the
  48.     // durable name (that is, as if you were passing the
  49.     // nats.Durable(<queue group name>) option.
  50.     // It is recommended to use nats.Bind() or nats.Durable()
  51.     // and preferably create the JetStream consumer beforehand
  52.     // (using js.AddConsumer) so that the JS consumer is not
  53.     // deleted on an Unsubscribe() or Drain() when the member
  54.     // that created the consumer goes away first.
  55.     // Check Godoc for the QueueSubscribe() API for more details.
  56.     js.QueueSubscribe("foo", "group", func(msg *nats.Msg) {
  57.         msg.Ack()
  58.     }, nats.ManualAck())
  60.     // Subscriber to consume messages synchronously.
  61.     sub, _ := js.SubscribeSync("foo")
  62.     msg, _ := sub.NextMsg(2 * time.Second)
  63.     msg.Ack()
  65.     // We can add a member to the group, with this member using
  66.     // the synchronous version of the QueueSubscribe.
  67.     sub, _ = js.QueueSubscribeSync("foo", "group")
  68.     msg, _ = sub.NextMsg(2 * time.Second)
  69.     msg.Ack()
  71.     // ChanSubscribe
  72.     msgCh := make(chan *nats.Msg, 8192)
  73.     sub, _ = js.ChanSubscribe("foo", msgCh)
  75.     select {
  76.     case msg := <-msgCh:
  77.         fmt.Println("[Received]", msg)
  78.     case <-time.After(1 * time.Second):
  79.     }
  80. }

{% endtab %} {% tab title=”Java” %}

  1. package io.nats.examples.jetstream;
  3. import io.nats.client.*;
  4. import io.nats.client.api.PublishAck;
  5. import io.nats.examples.ExampleArgs;
  6. import io.nats.examples.ExampleUtils;
  8. import java.io.IOException;
  9. import java.nio.charset.StandardCharsets;
  10. import java.time.Duration;
  11. import java.util.ArrayList;
  12. import java.util.List;
  13. import java.util.concurrent.atomic.AtomicInteger;
  15. import static io.nats.examples.jetstream.NatsJsUtils.createStreamExitWhenExists;
  17. /**
  18.  * This example will demonstrate JetStream push subscribing using a durable consumer and a queue
  19.  */
  20. public class NatsJsPushSubQueueDurable {
  21.     static final String usageString =
  22.         "\nUsage: java -cp <classpath> NatsJsPushSubQueueDurable [-s server] [-strm stream] [-sub subject] [-q queue] [-dur durable] [-mcnt msgCount] [-scnt subCount]"
  23.             + "\n\nDefault Values:"
  24.             + "\n   [-strm stream]   qdur-stream"
  25.             + "\n   [-sub subject]   qdur-subject"
  26.             + "\n   [-q queue]       qdur-queue"
  27.             + "\n   [-dur durable]   qdur-durable"
  28.             + "\n   [-mcnt msgCount] 100"
  29.             + "\n   [-scnt subCount] 5"
  30.             + "\n\nUse tls:// or opentls:// to require tls, via the Default SSLContext\n"
  31.             + "\nSet the environment variable NATS_NKEY to use challenge response authentication by setting a file containing your private key.\n"
  32.             + "\nSet the environment variable NATS_CREDS to use JWT/NKey authentication by setting a file containing your user creds.\n"
  33.             + "\nUse the URL in the -s server parameter for user/pass/token authentication.\n";
  35.     public static void main(String[] args) {
  36.         ExampleArgs exArgs = ExampleArgs.builder("Push Subscribe, Durable Consumer, Queue", args, usageString)
  37.                 .defaultStream("qdur-stream")
  38.                 .defaultSubject("qdur-subject")
  39.                 .defaultQueue("qdur-queue")
  40.                 .defaultDurable("qdur-durable")
  41.                 .defaultMsgCount(100)
  42.                 .defaultSubCount(5)
  43.                 .build();
  45.         try (Connection nc = Nats.connect(ExampleUtils.createExampleOptions(exArgs.server, true))) {
  47.             // Create a JetStreamManagement context.
  48.             JetStreamManagement jsm = nc.jetStreamManagement();
  50.             // Use the utility to create a stream stored in memory.
  51.             createStreamExitWhenExists(jsm, exArgs.stream, exArgs.subject);
  53.             // Create our JetStream context
  54.             JetStream js = nc.jetStream();
  56.             System.out.println();
  58.             // Setup the subscribers
  59.             // - the PushSubscribeOptions can be re-used since all the subscribers are the same
  60.             // - use a concurrent integer to track all the messages received
  61.             // - have a list of subscribers and threads so I can track them
  62.             PushSubscribeOptions pso = PushSubscribeOptions.builder().durable(exArgs.durable).build();
  63.             AtomicInteger allReceived = new AtomicInteger();
  64.             List<JsQueueSubscriber> subscribers = new ArrayList<>();
  65.             List<Thread> subThreads = new ArrayList<>();
  66.             for (int id = 1; id <= exArgs.subCount; id++) {
  67.                 // setup the subscription
  68.                 JetStreamSubscription sub = js.subscribe(exArgs.subject, exArgs.queue, pso);
  69.                 // create and track the runnable
  70.                 JsQueueSubscriber qs = new JsQueueSubscriber(id, exArgs, js, sub, allReceived);
  71.                 subscribers.add(qs);
  72.                 // create, track and start the thread
  73.                 Thread t = new Thread(qs);
  74.                 subThreads.add(t);
  75.                 t.start();
  76.             }
  77.             nc.flush(Duration.ofSeconds(1)); // flush outgoing communication with/to the server
  79.             // create and start the publishing
  80.             Thread pubThread = new Thread(new JsPublisher(js, exArgs));
  81.             pubThread.start();
  83.             // wait for all threads to finish
  84.             pubThread.join();
  85.             for (Thread t : subThreads) {
  86.                 t.join();
  87.             }
  89.             // report
  90.             for (JsQueueSubscriber qs : subscribers) {
  91.                 qs.report();
  92.             }
  94.             System.out.println();
  96.             // delete the stream since we are done with it.
  97.             jsm.deleteStream(exArgs.stream);
  98.         }
  99.         catch (Exception e) {
  100.             e.printStackTrace();
  101.         }
  102.     }
  104.     static class JsPublisher implements Runnable {
  105.         JetStream js;
  106.         ExampleArgs exArgs;
  108.         public JsPublisher(JetStream js, ExampleArgs exArgs) {
  109.             this.js = js;
  110.             this.exArgs = exArgs;
  111.         }
  113.         @Override
  114.         public void run() {
  115.             for (int x = 1; x <= exArgs.msgCount; x++) {
  116.                 try {
  117.                     PublishAck pa = js.publish(exArgs.subject, ("Data # " + x).getBytes(StandardCharsets.US_ASCII));
  118.                 } catch (IOException | JetStreamApiException e) {
  119.                     // something pretty wrong here
  120.                     e.printStackTrace();
  121.                     System.exit(-1);
  122.                 }
  123.             }
  124.         }
  125.     }
  127.     static class JsQueueSubscriber implements Runnable {
  128.         int id;
  129.         int thisReceived;
  130.         List<String> datas;
  132.         ExampleArgs exArgs;
  133.         JetStream js;
  134.         JetStreamSubscription sub;
  135.         AtomicInteger allReceived;
  137.         public JsQueueSubscriber(int id, ExampleArgs exArgs, JetStream js, JetStreamSubscription sub, AtomicInteger allReceived) {
  138.             this.id = id;
  139.             thisReceived = 0;
  140.             datas = new ArrayList<>();
  141.             this.exArgs = exArgs;
  142.             this.js = js;
  143.             this.sub = sub;
  144.             this.allReceived = allReceived;
  145.         }
  147.         public void report() {
  148.             System.out.printf("Sub # %d handled %d messages.\n", id, thisReceived);
  149.         }
  151.         @Override
  152.         public void run() {
  153.             while (allReceived.get() < exArgs.msgCount) {
  154.                 try {
  155.                     Message msg = sub.nextMessage(Duration.ofMillis(500));
  156.                     while (msg != null) {
  157.                         thisReceived++;
  158.                         allReceived.incrementAndGet();
  159.                         String data = new String(msg.getData(), StandardCharsets.US_ASCII);
  160.                         datas.add(data);
  161.                         System.out.printf("QS # %d message # %d %s\n", id, thisReceived, data);
  162.                         msg.ack();
  164.                         msg = sub.nextMessage(Duration.ofMillis(500));
  165.                     }
  166.                 } catch (InterruptedException e) {
  167.                     // just try again
  168.                 }
  169.             }
  170.             System.out.printf("QS # %d completed.\n", id);
  171.         }
  172.     }
  173. }

{% endtab %} {% tab title=”JavaScript” %}

  1. import { AckPolicy, connect } from "../../src/mod.ts";
  2. import { nuid } from "../../nats-base-client/nuid.ts";
  4. const nc = await connect();
  6. // create a regular subscription - this is plain nats
  7. const sub = nc.subscribe("my.messages", { max: 5 });
  8. const done = (async () => {
  9.   for await (const m of sub) {
  10.     console.log(m.subject);
  11.     m.respond();
  12.   }
  13. })();
  15. const jsm = await nc.jetstreamManager();
  16. const stream = nuid.next();
  17. const subj = nuid.next();
  18. await jsm.streams.add({ name: stream, subjects: [`${subj}.>`] });
  20. // create a consumer that delivers to the subscription
  21. await jsm.consumers.add(stream, {
  22.   ack_policy: AckPolicy.Explicit,
  23.   deliver_subject: "my.messages",
  24. });
  26. // publish some old nats messages
  27. nc.publish(`${subj}.A`);
  28. nc.publish(`${subj}.B`);
  29. nc.publish(`${subj}.C`);
  30. nc.publish(`${subj}.D.A`);
  31. nc.publish(`${subj}.F.A.B`);
  33. await done;
  34. await nc.close();

{% endtab %} {% tab title=”Python” %}

  1. import asyncio
  3. import nats
  4. from nats.errors import TimeoutError
  7. async def main():
  8.     nc = await nats.connect("localhost")
  10.     # Create JetStream context.
  11.     js = nc.jetstream()
  13.     # Persist messages on 'foo's subject.
  14.     await js.add_stream(name="sample-stream", subjects=["foo"])
  16.     for i in range(0, 10):
  17.         ack = await js.publish("foo", f"hello world: {i}".encode())
  18.         print(ack)
  20.     # Create pull based consumer on 'foo'.
  21.     psub = await js.pull_subscribe("foo", "psub")
  23.     # Fetch and ack messagess from consumer.
  24.     for i in range(0, 10):
  25.         msgs = await psub.fetch(1)
  26.         for msg in msgs:
  27.             print(msg)
  29.     # Create single push based subscriber that is durable across restarts.
  30.     sub = await js.subscribe("foo", durable="myapp")
  31.     msg = await sub.next_msg()
  32.     await msg.ack()
  34.     # Create deliver group that will be have load balanced messages.
  35.     async def qsub_a(msg):
  36.         print("QSUB A:", msg)
  37.         await msg.ack()
  39.     async def qsub_b(msg):
  40.         print("QSUB B:", msg)
  41.         await msg.ack()
  42.     await js.subscribe("foo", "workers", cb=qsub_a)
  43.     await js.subscribe("foo", "workers", cb=qsub_b)
  45.     for i in range(0, 10):
  46.         ack = await js.publish("foo", f"hello world: {i}".encode())
  47.         print("\t", ack)
  49.     await nc.close()
  51. if __name__ == '__main__':
  52.     asyncio.run(main())

{% endtab %} {% tab title=”C” %}

  1. #include "examples.h"
  3. static const char *usage = ""\
  4. "-gd            use global message delivery thread pool\n" \
  5. "-sync          receive synchronously (default is asynchronous)\n" \
  6. "-pull          use pull subscription\n" \
  7. "-fc            enable flow control\n" \
  8. "-count         number of expected messages\n";
  10. static void
  11. onMsg(natsConnection *nc, natsSubscription *sub, natsMsg *msg, void *closure)
  12. {
  13.     if (print)
  14.         printf("Received msg: %s - %.*s\n",
  15.                natsMsg_GetSubject(msg),
  16.                natsMsg_GetDataLength(msg),
  17.                natsMsg_GetData(msg));
  19.     if (start == 0)
  20.         start = nats_Now();
  22.     // We should be using a mutex to protect those variables since
  23.     // they are used from the subscription's delivery and the main
  24.     // threads. For demo purposes, this is fine.
  25.     if (++count == total)
  26.         elapsed = nats_Now() - start;
  28.     // Since this is auto-ack callback, we don't need to ack here.
  29.     natsMsg_Destroy(msg);
  30. }
  32. static void
  33. asyncCb(natsConnection *nc, natsSubscription *sub, natsStatus err, void *closure)
  34. {
  35.     printf("Async error: %u - %s\n", err, natsStatus_GetText(err));
  37.     natsSubscription_GetDropped(sub, (int64_t*) &dropped);
  38. }
  40. int main(int argc, char **argv)
  41. {
  42.     natsConnection      *conn  = NULL;
  43.     natsStatistics      *stats = NULL;
  44.     natsOptions         *opts  = NULL;
  45.     natsSubscription    *sub   = NULL;
  46.     natsMsg             *msg   = NULL;
  47.     jsCtx               *js    = NULL;
  48.     jsErrCode           jerr   = 0;
  49.     jsOptions           jsOpts;
  50.     jsSubOptions        so;
  51.     natsStatus          s;
  52.     bool                delStream = false;
  54.     opts = parseArgs(argc, argv, usage);
  56.     printf("Created %s subscription on '%s'.\n",
  57.         (pull ? "pull" : (async ? "asynchronous" : "synchronous")), subj);
  59.     s = natsOptions_SetErrorHandler(opts, asyncCb, NULL);
  61.     if (s == NATS_OK)
  62.         s = natsConnection_Connect(&conn, opts);
  64.     if (s == NATS_OK)
  65.         s = jsOptions_Init(&jsOpts);
  67.     if (s == NATS_OK)
  68.         s = jsSubOptions_Init(&so);
  69.     if (s == NATS_OK)
  70.     {
  71.         so.Stream = stream;
  72.         so.Consumer = durable;
  73.         if (flowctrl)
  74.         {
  75.             so.Config.FlowControl = true;
  76.             so.Config.Heartbeat = (int64_t)1E9;
  77.         }
  78.     }
  80.     if (s == NATS_OK)
  81.         s = natsConnection_JetStream(&js, conn, &jsOpts);
  83.     if (s == NATS_OK)
  84.     {
  85.         jsStreamInfo    *si = NULL;
  87.         // First check if the stream already exists.
  88.         s = js_GetStreamInfo(&si, js, stream, NULL, &jerr);
  89.         if (s == NATS_NOT_FOUND)
  90.         {
  91.             jsStreamConfig  cfg;
  93.             // Since we are the one creating this stream, we can delete at the end.
  94.             delStream = true;
  96.             // Initialize the configuration structure.
  97.             jsStreamConfig_Init(&cfg);
  98.             cfg.Name = stream;
  99.             // Set the subject
  100.             cfg.Subjects = (const char*[1]){subj};
  101.             cfg.SubjectsLen = 1;
  102.             // Make it a memory stream.
  103.             cfg.Storage = js_MemoryStorage;
  104.             // Add the stream,
  105.             s = js_AddStream(&si, js, &cfg, NULL, &jerr);
  106.         }
  107.         if (s == NATS_OK)
  108.         {
  109.             printf("Stream %s has %" PRIu64 " messages (%" PRIu64 " bytes)\n",
  110.                 si->Config->Name, si->State.Msgs, si->State.Bytes);
  112.             // Need to destroy the returned stream object.
  113.             jsStreamInfo_Destroy(si);
  114.         }
  115.     }
  117.     if (s == NATS_OK)
  118.     {
  119.         if (pull)
  120.             s = js_PullSubscribe(&sub, js, subj, durable, &jsOpts, &so, &jerr);
  121.         else if (async)
  122.             s = js_Subscribe(&sub, js, subj, onMsg, NULL, &jsOpts, &so, &jerr);
  123.         else
  124.             s = js_SubscribeSync(&sub, js, subj, &jsOpts, &so, &jerr);
  125.     }
  126.     if (s == NATS_OK)
  127.         s = natsSubscription_SetPendingLimits(sub, -1, -1);
  129.     if (s == NATS_OK)
  130.         s = natsStatistics_Create(&stats);
  132.     if ((s == NATS_OK) && pull)
  133.     {
  134.         natsMsgList list;
  135.         int         i;
  137.         for (count = 0; (s == NATS_OK) && (count < total); )
  138.         {
  139.             s = natsSubscription_Fetch(&list, sub, 1024, 5000, &jerr);
  140.             if (s != NATS_OK)
  141.                 break;
  143.             if (start == 0)
  144.                 start = nats_Now();
  146.             count += (int64_t) list.Count;
  147.             for (i=0; (s == NATS_OK) && (i<list.Count); i++)
  148.                 s = natsMsg_Ack(list.Msgs[i], &jsOpts);
  150.             natsMsgList_Destroy(&list);
  151.         }
  152.     }
  153.     else if ((s == NATS_OK) && async)
  154.     {
  155.         while (s == NATS_OK)
  156.         {
  157.             if (count + dropped == total)
  158.                 break;
  160.             nats_Sleep(1000);
  161.         }
  162.     }
  163.     else if (s == NATS_OK)
  164.     {
  165.         for (count = 0; (s == NATS_OK) && (count < total); count++)
  166.         {
  167.             s = natsSubscription_NextMsg(&msg, sub, 5000);
  168.             if (s != NATS_OK)
  169.                 break;
  171.             if (start == 0)
  172.                 start = nats_Now();
  174.             s = natsMsg_Ack(msg, &jsOpts);
  175.             natsMsg_Destroy(msg);
  176.         }
  177.     }
  179.     if (s == NATS_OK)
  180.     {
  181.         printStats(STATS_IN|STATS_COUNT, conn, sub, stats);
  182.         printPerf("Received");
  183.     }
  184.     if (s == NATS_OK)
  185.     {
  186.         jsStreamInfo *si = NULL;
  188.         // Let's report some stats after the run
  189.         s = js_GetStreamInfo(&si, js, stream, NULL, &jerr);
  190.         if (s == NATS_OK)
  191.         {
  192.             printf("\nStream %s has %" PRIu64 " messages (%" PRIu64 " bytes)\n",
  193.                 si->Config->Name, si->State.Msgs, si->State.Bytes);
  195.             jsStreamInfo_Destroy(si);
  196.         }
  197.         if (delStream)
  198.         {
  199.             printf("\nDeleting stream %s: ", stream);
  200.             s = js_DeleteStream(js, stream, NULL, &jerr);
  201.             if (s == NATS_OK)
  202.                 printf("OK!");
  203.             printf("\n");
  204.         }
  205.     }
  206.     else
  207.     {
  208.         printf("Error: %u - %s - jerr=%u\n", s, natsStatus_GetText(s), jerr);
  209.         nats_PrintLastErrorStack(stderr);
  210.     }
  212.     // Destroy all our objects to avoid report of memory leak
  213.     jsCtx_Destroy(js);
  214.     natsStatistics_Destroy(stats);
  215.     natsSubscription_Destroy(sub);
  216.     natsConnection_Destroy(conn);
  217.     natsOptions_Destroy(opts);
  219.     // To silence reports of memory still in used with valgrind
  220.     nats_Close();.
  222.     return 0;
  223. }

{% endtab %} {% endtabs %}

Delivery reliability

JetStream consumers can ensure not just the reliability of message delivery but also the reliability of the processing of the messages, even in the face of client application or downstream failures. It does so by using message level acknowledgements and message re-deliveries.

Consumers have an Acknowledgement Policy specifying the level of reliability required. In increasing order of reliability the available policies are: ‘none’ for no application level acknowledgements, ‘all’ where acknowledging a specific message also implicitly acknowledges all previous messages in the stream, and ‘explicit’ where each message must be individually acknowledged.

When the consumer is set to require explicit acknowledgements the client applications are able to use more than one kind of acknowledgement to indicate successful (or not) reception and processing of the messages being received from the consumer.

Applications can:

  • Acknowledge the successfull processing of a message (Ack()).
  • Acknowledge the successfull processing of a message and request an acknowledgement of the reception of the acknowledgement by the consumer (AckSync()).
  • Indicate that the processing is still in progress and more time is needed (inProgress()).
  • Negatively acknowledge a message, indicating that the client application is currently (temporarily) unable to process the message and that the consumer should attempt to re-deliver it (Nak()).
  • Terminate a message (typically, because there is a problem with the data inside the message such that the client application is never going to be able to process it), indicating that the consumer should not attempt to re-deliver the message (Term()).

After a message is sent from the consumer to a subscribing client application by the server an ‘AckWait’ timer is started. This timer is deleted when either a positive (Ack()) or a termination (Term()) acknowledgement is received from the client application. The timer gets reset upon reception of an in-progress (inProgress()) acknowledgement.

If at the end of a period of time no acknowledgement has been received from the client application, the server will attempt to re-deliver the message. If there is more than one client application instance subscribing to the consumer, there is no guarantee that the re-delivery would be to any particular client instance.

You can control the timing of re-deliveries using either the single AckWait duration attribute of the consumer, or as a sequence of durations in the BackOff attribute (which overrides AckWait).

You can also control the timing of re-deliveries when messages are negatively acknowledged with Nak(), by passing a nakDelay() option (or using NakWithDelay()), otherwise the re-delivery attempt will happen right after the reception of the Nak by the server.

“Dead Letter Queues” type functionality

You can set a maximum number of delivery attempts using the consumer’s MaxDeliver setting.

Whenever a message reaches its maximum number of delivery attempts an advisory message is published on the $JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.<STREAM>.<CONSUMER> subject. The advisory message’s payload (use nats schema info io.nats.jetstream.advisory.v1.max_deliver for specific information) contains a stream_seq field that contains the sequence number of the message in the stream.

Similarly, whenever a client application terminates delivery attempts for the message using AckTerm an advisory message is published on the $JS.EVENT.ADVISORY.CONSUMER.MSG_TERMINATED.<STREAM>.<CONSUMER> subject, and its payload (see nats schema info io.nats.jetstream.advisory.v1.terminated) contains a stream_seq field.

You can leverage those advisory messages to implement “Dead Letter Queue” (DLQ) types of functionalities. For example:

  • If you only need to know about each time a message is ‘dead’ (considered un-re-deliverable by the consumer), then listening to the advisories is enough.
  • If you also need to have access to the message in question then you can use the message’s sequence number included in the advisory to retrieve that specific message by sequence number from the stream. If a message reaches its maximum level of delivery attempts, it will still stay in the stream until it is manually deleted or manually acknowledged.