Managing Streams and consumers

Streams and durable consumers can be defined administratively outside the application (typically using the NATS CLI Tool) in which case the application only needs to know about the well-known names of the durable consumers it wants to use. But you can also manage streams and consumers programmatically.

Common stream management operations are:

  • Add a stream. Adding a stream is an idempotent function, which means that if a stream does not exist, it will be created, and if a stream already exists, then the add operation will succeed only if the existing stream matches exactly the attributes specified in the ‘add’ call.
  • Delete a stream.
  • Purge a stream (delete all the messages stored in the stream)
  • Get or remove a specific message from a stream by sequence number
  • Add or update (or delete) a consumer
  • Get info and statistics on streams/consumers/account. Get/remove/get information on individual messages stored in a stream.

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

  1. func ExampleJetStreamManager() {
  2.     nc, _ := nats.Connect("localhost")
  4.     js, _ := nc.JetStream()
  6.     // Create a stream
  7.     js.AddStream(&nats.StreamConfig{
  8.         Name:     "example-stream",
  9.         Subjects: []string{"example-subject"},
  10.         MaxBytes: 1024,
  11.     })
  13.     // Update a stream
  14.     js.UpdateStream(&nats.StreamConfig{
  15.         Name:     "example-stream",
  16.         MaxBytes: 2048,
  17.     })
  19.     // Create a durable consumer
  20.     js.AddConsumer("example-stream", &nats.ConsumerConfig{
  21.         Durable: "example-consumer-name",
  22.     })
  24.     // Get information about all streams (with Context JSOpt)
  25.     ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
  26.     defer cancel()
  27.     for info := range js.StreamsInfo(nats.Context(ctx)) {
  28.         fmt.Println("stream name: ", info.Config.Name)
  29.     }
  31.     // Get information about all consumers (with MaxWait JSOpt)
  32.     for info := range js.ConsumersInfo("example-stream", nats.MaxWait(10*time.Second)) {
  33.         fmt.Println("consumer name: ", info.Name)
  34.     }
  36.     // Delete a consumer
  37.     js.DeleteConsumer("example-stream", "example-consumer-name")
  39.     // Delete a stream
  40.     js.DeleteStream("example-stream")
  41. }

{% endtab %}

{% tab title=”Java” %}

  1. try (Connection nc = Nats.connect("localhost")) {
  2.     JetStreamManagement jsm = nc.jetStreamManagement();
  4.     // Create a stream
  5.     StreamInfo si = jsm.addStream(StreamConfiguration.builder()
  6.         .name("example-stream")
  7.         .subjects("example-subject")
  8.         .maxBytes(1024)
  9.         .build());
  10.     StreamConfiguration config = si.getConfiguration();
  11.     System.out.println("stream name: " + config.getName() + ", max_bytes: " + config.getMaxBytes());
  13.     // Update a stream
  14.     si = jsm.updateStream(StreamConfiguration.builder()
  15.         .name("example-stream")
  16.         .maxBytes(2048)
  17.         .build());
  18.     config = si.getConfiguration();
  19.     System.out.println("stream name: " + config.getName() + ", max_bytes: " + config.getMaxBytes());
  21.     // Create a durable consumer
  22.     jsm.createConsumer("example-stream", ConsumerConfiguration.builder()
  23.         .durable("example-consumer-name")
  24.         .build());
  26.     // Get information about all streams
  27.     List<StreamInfo> streams = jsm.getStreams();
  28.     for (StreamInfo info : streams) {
  29.         System.out.println("stream name: " + info.getConfiguration().getName());
  30.     }
  32.     // Get information about all consumers
  33.     List<ConsumerInfo> consumers = jsm.getConsumers("example-stream");
  34.     for (ConsumerInfo ci : consumers) {
  35.         System.out.println("consumer name: " + ci.getName());
  36.     }
  38.     // Delete a consumer
  39.     jsm.deleteConsumer("example-stream", "example-consumer-name");
  41.     // Delete a stream
  42.     jsm.deleteStream("example-stream");
  43. }

{% endtab %}

{% tab title=”JavaScript” %}

  1. import { AckPolicy, connect, Empty } from "../../src/mod.ts";
  3. const nc = await connect();
  4. const jsm = await nc.jetstreamManager();
  6. // list all the streams, the `next()` function
  7. // retrieves a paged result.
  8. const streams = await jsm.streams.list().next();
  9. streams.forEach((si) => {
  10.     console.log(si);
  11. });
  13. // add a stream
  14. const stream = "mystream";
  15. const subj = `mystream.*`;
  16. await jsm.streams.add({ name: stream, subjects: [subj] });
  18. // publish a reg nats message directly to the stream
  19. for (let i = 0; i < 10; i++) {
  20.     nc.publish(`${subj}.a`, Empty);
  21. }
  23. // find a stream that stores a specific subject:
  24. const name = await jsm.streams.find("mystream.A");
  26. // retrieve info about the stream by its name
  27. const si = await jsm.streams.info(name);
  29. // update a stream configuration
  30. si.config.subjects?.push("a.b");
  31. await jsm.streams.update(name, si.config);
  33. // get a particular stored message in the stream by sequence
  34. // this is not associated with a consumer
  35. const sm = await jsm.streams.getMessage(stream, { seq: 1 });
  36. console.log(sm.seq);
  38. // delete the 5th message in the stream, securely erasing it
  39. await jsm.streams.deleteMessage(stream, 5);
  41. // purge all messages in the stream, the stream itself
  42. // remains.
  43. await jsm.streams.purge(stream);
  45. // purge all messages with a specific subject (filter can be a wildcard)
  46. await jsm.streams.purge(stream, { filter: "a.b" });
  48. // purge messages with a specific subject keeping some messages
  49. await jsm.streams.purge(stream, { filter: "a.c", keep: 5 });
  51. // purge all messages with upto (not including seq)
  52. await jsm.streams.purge(stream, { seq: 100 });
  54. // purge all messages with upto sequence that have a matching subject
  55. await jsm.streams.purge(stream, { filter: "a.d", seq: 100 });
  57. // list all consumers for a stream:
  58. const consumers = await jsm.consumers.list(stream).next();
  59. consumers.forEach((ci) => {
  60.     console.log(ci);
  61. });
  63. // add a new durable pull consumer
  64. await jsm.consumers.add(stream, {
  65.     durable_name: "me",
  66.     ack_policy: AckPolicy.Explicit,
  67. });
  69. // retrieve a consumer's configuration
  70. const ci = await jsm.consumers.info(stream, "me");
  71. console.log(ci);
  73. // delete a particular consumer
  74. await jsm.consumers.delete(stream, "me");

{% 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.     await nc.close()
  17. if __name__ == '__main__':
  18.     asyncio.run(main())

{% endtab %}

{% tab title=”C# V1” %}

  1. using (IConnection nc = new ConnectionFactory().CreateConnection("nats://localhost:4222"))
  2. {
  3.     IJetStreamManagement jsm = nc.CreateJetStreamManagementContext();
  5.     // Create a stream
  6.     StreamInfo si = jsm.AddStream(StreamConfiguration.Builder()
  7.         .WithName("example-stream")
  8.         .WithSubjects("example-subject")
  9.         .WithMaxBytes(1024)
  10.         .Build());
  11.     Console.WriteLine($"stream name: {si.Config.Name}, max_bytes: {si.Config.MaxBytes}");
  13.     // Update a stream
  14.     si = jsm.UpdateStream(StreamConfiguration.Builder()
  15.         .WithName("example-stream")
  16.         .WithMaxBytes(2048)
  17.         .Build());
  18.     Console.WriteLine($"stream name: {si.Config.Name}, max_bytes: {si.Config.MaxBytes}");
  20.     // Create a durable consumer
  21.     jsm.CreateConsumer("example-stream", ConsumerConfiguration.Builder()
  22.         .WithDurable("example-consumer-name")
  23.         .Build());
  25.     // Get information about all streams
  26.     IList<StreamInfo> streams = jsm.GetStreams();
  27.     foreach (StreamInfo info in streams) {
  28.         Console.WriteLine($"stream name: {info.Config.Name}");
  29.     }
  31.     // Get information about all consumers
  32.     IList<ConsumerInfo> consumers = jsm.GetConsumers("example-stream");
  33.     foreach (ConsumerInfo ci in consumers) {
  34.         Console.WriteLine($"consumer name: {ci.Name}");
  35.     }
  37.     // Delete a consumer
  38.     jsm.DeleteConsumer("example-stream", "example-consumer-name");
  40.     // Delete a stream
  41.     jsm.DeleteStream("example-stream");
  42. }

{% endtab %}

{% tab title=”C” %}

  1. #include "examples.h"
  3. static const char *usage = ""\
  4. "-stream        stream name (default is 'foo')\n" \
  5. "-txt           text to send (default is 'hello')\n" \
  6. "-count         number of messages to send\n" \
  7. "-sync          publish synchronously (default is async)\n";
  9. static void
  10. _jsPubErr(jsCtx *js, jsPubAckErr *pae, void *closure)
  11. {
  12.     int *errors = (int*) closure;
  14.     printf("Error: %u - Code: %u - Text: %s\n", pae->Err, pae->ErrCode, pae->ErrText);
  15.     printf("Original message: %.*s\n", natsMsg_GetDataLength(pae->Msg), natsMsg_GetData(pae->Msg));
  17.     *errors = (*errors + 1);
  19.     // If we wanted to resend the original message, we would do something like that:
  20.     //
  21.     // js_PublishMsgAsync(js, &(pae->Msg), NULL);
  22.     //
  23.     // Note that we use `&(pae->Msg)` so that the library set it to NULL if it takes
  24.     // ownership, and the library will not destroy the message when this callback returns.
  26.     // No need to destroy anything, everything is handled by the library.
  27. }
  29. int main(int argc, char **argv)
  30. {
  31.     natsConnection      *conn  = NULL;
  32.     natsStatistics      *stats = NULL;
  33.     natsOptions         *opts  = NULL;
  34.     jsCtx               *js    = NULL;
  35.     jsOptions           jsOpts;
  36.     jsErrCode           jerr   = 0;
  37.     natsStatus          s;
  38.     int                 dataLen=0;
  39.     volatile int        errors = 0;
  40.     bool                delStream = false;
  42.     opts = parseArgs(argc, argv, usage);
  43.     dataLen = (int) strlen(payload);
  45.     s = natsConnection_Connect(&conn, opts);
  47.     if (s == NATS_OK)
  48.         s = jsOptions_Init(&jsOpts);
  50.     if (s == NATS_OK)
  51.     {
  52.         if (async)
  53.         {
  54.             jsOpts.PublishAsync.ErrHandler           = _jsPubErr;
  55.             jsOpts.PublishAsync.ErrHandlerClosure    = (void*) &errors;
  56.         }
  57.         s = natsConnection_JetStream(&js, conn, &jsOpts);
  58.     }
  60.     if (s == NATS_OK)
  61.     {
  62.         jsStreamInfo    *si = NULL;
  64.         // First check if the stream already exists.
  65.         s = js_GetStreamInfo(&si, js, stream, NULL, &jerr);
  66.         if (s == NATS_NOT_FOUND)
  67.         {
  68.             jsStreamConfig  cfg;
  70.             // Since we are the one creating this stream, we can delete at the end.
  71.             delStream = true;
  73.             // Initialize the configuration structure.
  74.             jsStreamConfig_Init(&cfg);
  75.             cfg.Name = stream;
  76.             // Set the subject
  77.             cfg.Subjects = (const char*[1]){subj};
  78.             cfg.SubjectsLen = 1;
  79.             // Make it a memory stream.
  80.             cfg.Storage = js_MemoryStorage;
  81.             // Add the stream,
  82.             s = js_AddStream(&si, js, &cfg, NULL, &jerr);
  83.         }
  84.         if (s == NATS_OK)
  85.         {
  86.             printf("Stream %s has %" PRIu64 " messages (%" PRIu64 " bytes)\n",
  87.                 si->Config->Name, si->State.Msgs, si->State.Bytes);
  89.             // Need to destroy the returned stream object.
  90.             jsStreamInfo_Destroy(si);
  91.         }
  92.     }
  94.     if (s == NATS_OK)
  95.         s = natsStatistics_Create(&stats);
  97.     if (s == NATS_OK)
  98.     {
  99.         printf("\nSending %" PRId64 " messages to subject '%s'\n", total, stream);
  100.         start = nats_Now();
  101.     }
  103.     for (count = 0; (s == NATS_OK) && (count < total); count++)
  104.     {
  105.         if (async)
  106.             s = js_PublishAsync(js, subj, (const void*) payload, dataLen, NULL);
  107.         else
  108.         {
  109.             jsPubAck *pa = NULL;
  111.             s = js_Publish(&pa, js, subj, (const void*) payload, dataLen, NULL, &jerr);
  112.             if (s == NATS_OK)
  113.             {
  114.                 if (pa->Duplicate)
  115.                     printf("Got a duplicate message! Sequence=%" PRIu64 "\n", pa->Sequence);
  117.                 jsPubAck_Destroy(pa);
  118.             }
  119.         }
  120.     }
  122.     if ((s == NATS_OK) && async)
  123.     {
  124.         jsPubOptions    jsPubOpts;
  126.         jsPubOptions_Init(&jsPubOpts);
  127.         // Let's set it to 30 seconds, if getting "Timeout" errors,
  128.         // this may need to be increased based on the number of messages
  129.         // being sent.
  130.         jsPubOpts.MaxWait = 30000;
  131.         s = js_PublishAsyncComplete(js, &jsPubOpts);
  132.         if (s == NATS_TIMEOUT)
  133.         {
  134.             // Let's get the list of pending messages. We could resend,
  135.             // etc, but for now, just destroy them.
  136.             natsMsgList list;
  138.             js_PublishAsyncGetPendingList(&list, js);
  139.             natsMsgList_Destroy(&list);
  140.         }
  141.     }
  143.     if (s == NATS_OK)
  144.     {
  145.         jsStreamInfo *si = NULL;
  147.         elapsed = nats_Now() - start;
  148.         printStats(STATS_OUT, conn, NULL, stats);
  149.         printPerf("Sent");
  151.         if (errors != 0)
  152.             printf("There were %d asynchronous errors\n", errors);
  154.         // Let's report some stats after the run
  155.         s = js_GetStreamInfo(&si, js, stream, NULL, &jerr);
  156.         if (s == NATS_OK)
  157.         {
  158.             printf("\nStream %s has %" PRIu64 " messages (%" PRIu64 " bytes)\n",
  159.                 si->Config->Name, si->State.Msgs, si->State.Bytes);
  161.             jsStreamInfo_Destroy(si);
  162.         }
  163.     }
  164.     if (delStream && (js != NULL))
  165.     {
  166.         printf("\nDeleting stream %s: ", stream);
  167.         s = js_DeleteStream(js, stream, NULL, &jerr);
  168.         if (s == NATS_OK)
  169.             printf("OK!");
  170.         printf("\n");
  171.     }
  172.     if (s != NATS_OK)
  173.     {
  174.         printf("Error: %u - %s - jerr=%u\n", s, natsStatus_GetText(s), jerr);
  175.         nats_PrintLastErrorStack(stderr);
  176.     }
  178.     // Destroy all our objects to avoid report of memory leak
  179.     jsCtx_Destroy(js);
  180.     natsStatistics_Destroy(stats);
  181.     natsConnection_Destroy(conn);
  182.     natsOptions_Destroy(opts);
  184.     // To silence reports of memory still in used with valgrind
  185.     nats_Close();
  187.     return 0;
  188. }
  189. ```
  191. </div>
  193. </div>
  195. </div>
  197. </div>

{% endtab %} {% endtabs %}