Draining Messages Before Disconnect

    Drain provides clients that use queue subscriptions with a way to bring down applications without losing any messages. A client can bring up a new queue member, drain and shut down the old queue member, all without losing messages sent to the old client. Without drain, there is the possibility of lost messages due to delivery timing.

    The libraries can provide drain on a connection or on a subscriber, or both.

    For a connection the process is essentially:

    1. Drain all subscriptions
    2. Stop new messages from being published
    3. Flush any remaining published messages
    4. Close

    The API for drain can generally be used instead of close:

    As an example of draining a connection:

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

    {% endtab %}

    {% tab title=”Java” %}

    3. // Use a latch to wait for a message to arrive
    4. CountDownLatch latch = new CountDownLatch(1);
    6. // Create a dispatcher and inline message handler
    7. Dispatcher d = nc.createDispatcher((msg) -> {
    8.     String str = new String(msg.getData(), StandardCharsets.UTF_8);
    9.     System.out.println(str);
    10.     latch.countDown();
    11. });
    13. // Subscribe
    14. d.subscribe("updates");
    16. // Wait for a message to come in
    17. latch.await();
    19. // Drain the connection, which will close it
    20. CompletableFuture<Boolean> drained = nc.drain(Duration.ofSeconds(10));
    22. // Wait for the drain to complete
    23. drained.get();

    {% endtab %}

    {% tab title=”JavaScript” %}

    1. const nc = await connect({ servers: "demo.nats.io" });
    2. const sub = nc.subscribe(createInbox(), () => {});
    3. nc.publish(sub.getSubject());
    4. await nc.drain();

    {% tab title=”Python” %}

    1. import asyncio
    2. from nats.aio.client import Client as NATS
    4. async def example(loop):
    5.     nc = NATS()
    7.     await nc.connect("nats://127.0.0.1:4222", loop=loop)
    9.     async def handler(msg):
    10.         print("[Received] ", msg)
    11.         await nc.publish(msg.reply, b'I can help')
    13.         # Can check whether client is in draining state
    14.         if nc.is_draining:
    15.             print("Connection is draining")
    17.     await nc.subscribe("help", "workers", cb=handler)
    18.     await nc.flush()
    20.     requests = []
    21.     for i in range(0, 10):
    22.         request = nc.request("help", b'help!', timeout=1)
    23.         requests.append(request)
    25.     # Wait for all the responses
    26.     responses = []
    27.     responses = await asyncio.gather(*requests)
    29.     # Gracefully close the connection.
    30.     await nc.drain()
    32.     print("Received {} responses".format(len(responses)))

    {% endtab %}

    {% tab title=”Ruby” %}

    {% 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.     // Add some delay while processing
    10.     nats_Sleep(200);
    12.     // Need to destroy the message!
    13.     natsMsg_Destroy(msg);
    14. }
    16. static void
    17. closeHandler(natsConnection *conn, void *closure)
    18. {
    19.     cond_variable cv = (cond_variable) closure;
    21.     notify_cond_variable(cv);
    22. }
    23. (...)
    26. natsConnection      *conn      = NULL;
    27. natsOptions         *opts      = NULL;
    28. natsSubscription    *sub       = NULL;
    29. natsStatus          s          = NATS_OK;
    30. cond_variable       cv         = new_cond_variable(); // some fictuous way to notify between threads.
    32. s = natsOptions_Create(&opts);
    33. if (s == NATS_OK)
    34.     // Setup a close handler and pass a reference to our condition variable.
    35.     s = natsOptions_SetClosedCB(opts, closeHandler, (void*) cv);
    36. if (s == NATS_OK)
    37.     s = natsConnection_Connect(&conn, opts);
    39. // Subscribe
    40. if (s == NATS_OK)
    41.     s = natsConnection_Subscribe(&sub, conn, "foo", onMsg, NULL);
    43. // Publish a message
    44. if (s == NATS_OK)
    45.     s = natsConnection_PublishString(conn, "foo", "hello");
    47. // Drain the connection, which will close it when done.
    48. if (s == NATS_OK)
    49.     s = natsConnection_Drain(conn);
    51. // Wait for the connection to be closed
    52. if (s == NATS_OK)
    53.     cond_variable_wait(cv);
    55. (...)
    57. // Destroy objects that were created
    58. natsSubscription_Destroy(sub);
    59. natsConnection_Destroy(conn);
    60. natsOptions_Destroy(opts);

    {% endtab %} {% endtabs %}

    The mechanics of drain for a subscription are simpler:

    1. Unsubscribe
    2. Process all cached or inflight messages
    3. Clean up

    The API for drain can generally be used instead of unsubscribe:

    {% 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.     done := sync.WaitGroup{}
    8.     done.Add(1)
    10.     count := 0
    11.     errCh := make(chan error, 1)
    13.     msgAfterDrain := "not this one"
    15.     // Just to not collide using the demo server with other users.
    16.     subject := nats.NewInbox()
    18.     // This callback will process each message slowly
    19.     sub, err := nc.Subscribe(subject, func(m *nats.Msg) {
    20.         if string(m.Data) == msgAfterDrain {
    21.             errCh <- fmt.Errorf("Should not have received this message")
    22.             return
    23.         }
    24.         time.Sleep(100 * time.Millisecond)
    25.         count++
    26.         if count == 2 {
    27.             done.Done()
    28.         }
    29.     })
    31.     // Send 2 messages
    32.     for i := 0; i < 2; i++ {
    33.         nc.Publish(subject, []byte("hello"))
    34.     }
    36.     // Call Drain on the subscription. It unsubscribes but
    37.     // wait for all pending messages to be processed.
    38.     if err := sub.Drain(); err != nil {
    39.         log.Fatal(err)
    40.     }
    42.     // Send one more message, this message should not be received
    43.     nc.Publish(subject, []byte(msgAfterDrain))
    45.     done.Wait()
    47.     // Now check that the 3rd message was not received
    49.     case e := <-errCh:
    50.         log.Fatal(e)
    51.     case <-time.After(200 * time.Millisecond):
    52.         // OK!
    53.     }

    {% endtab %}

    1. Connection nc = Nats.connect("nats://demo.nats.io:4222");
    3. // Use a latch to wait for a message to arrive
    4. CountDownLatch latch = new CountDownLatch(1);
    6. // Create a dispatcher and inline message handler
    7. Dispatcher d = nc.createDispatcher((msg) -> {
    8.     String str = new String(msg.getData(), StandardCharsets.UTF_8);
    9.     System.out.println(str);
    10.     latch.countDown();
    11. });
    13. // Subscribe
    14. d.subscribe("updates");
    16. // Wait for a message to come in
    17. latch.await();
    19. // Messages that have arrived will be processed
    20. CompletableFuture<Boolean> drained = d.drain(Duration.ofSeconds(10));
    22. // Wait for the drain to complete
    23. drained.get();
    25. // Close the connection
    26. nc.close();

    {% endtab %}

    {% tab title=”JavaScript” %}

    {% endtab %}

    {% tab title=”Python” %}

    1. import asyncio
    2. from nats.aio.client import Client as NATS
    4. async def example(loop):
    5.     nc = NATS()
    7.     await nc.connect("nats://127.0.0.1:4222", loop=loop)
    9.     async def handler(msg):
    10.         print("[Received] ", msg)
    11.         await nc.publish(msg.reply, b'I can help')
    13.         # Can check whether client is in draining state
    14.         if nc.is_draining:
    15.             print("Connection is draining")
    17.     sid = await nc.subscribe("help", "workers", cb=handler)
    18.     await nc.flush()
    20.     # Gracefully unsubscribe the subscription
    21.     await nc.drain(sid)

    {% endtab %}

    {% tab title=”Ruby” %}

    1. # There is currently no API to drain a single subscription, the whole connection can be drained though via NATS.drain

    {% endtab %}

    {% tab title=”C” %}

    1. natsConnection      *conn      = NULL;
    2. natsSubscription    *sub       = NULL;
    3. natsStatus          s          = NATS_OK;
    5. s = natsConnection_ConnectTo(&conn, NATS_DEFAULT_URL);
    7. // Subscribe
    8. if (s == NATS_OK)
    9.     s = natsConnection_Subscribe(&sub, conn, "foo", onMsg, NULL);
    11. // Publish 2 messages
    12. if (s == NATS_OK)
    13. {
    14.     int i;
    15.     for (i=0; (s == NATS_OK) && (i<2); i++)
    16.     {
    17.         s = natsConnection_PublishString(conn, "foo", "hello");
    18.     }
    19. }
    21. // Call Drain on the subscription. It unsubscribes but
    22. // wait for all pending messages to be processed.
    23. if (s == NATS_OK)
    24.     s = natsSubscription_Drain(sub);
    26. (...)
    28. // Destroy objects that were created
    29. natsConnection_Destroy(conn);

    {% endtab %} {% endtabs %}

    Because draining can involve messages flowing to the server, for a flush and asynchronous message processing, the timeout for drain should generally be higher than the timeout for a simple message request/reply or similar.