subscription.drain isn't waiting for buffered messages to be processed

[AlixBa]

Observed behavior

using subscription.drain(timeout) doesn't wait until the messages for this subscription in the buffer are processed to complete (and complete positively).

Expected behavior

subscription.drain(timeout) should return true when all buffered messages have been processed, or it should return an error if it's not intended to be used through subscription but through dispatcher.drain, or maybe it should be explained in the java doc?

Server and client version

server: nats:2.9.22-alpine
client: 2.22.0

Host environment

No response

Steps to reproduce

import cats.effect.IO
import cats.effect.IOApp
import io.nats.client.Nats
import io.nats.client.Options

import java.time.Duration

object Main2 extends IOApp.Simple {
  override def run: IO[Unit] = IO.delay {
    val natsUrl = "nats://localhost:4222"
    val options = Options.Builder().server(natsUrl).build()
    val connection = Nats.connect(options)

    val dispatcher = connection.createDispatcher()
    val subscription = dispatcher.subscribe(
      "subject",
      message => {
        println("received message " + String(message.getData()))
        Thread.sleep(1000)
      }
    )

    println("publishing 5 messages")
    connection.publish("subject", "1".getBytes())
    connection.publish("subject", "2".getBytes())
    connection.publish("subject", "3".getBytes())
    connection.publish("subject", "4".getBytes())
    connection.publish("subject", "5".getBytes())
    println("published all")

    Thread.sleep(1000)

    // works as intended with dispatcher.drain
    // but there might be some cases we don't want to drain the dispatcher as a whole
    val drained = subscription.drain(Duration.ofSeconds(30)).get()
    println("drained: " + drained)
    connection.drain(Duration.ofSeconds(30)).get()
  }
}

output

[116] publishing 5 messages
[116] published all
[116] received message 1
[116] received message 2
[116] drained: true

[scottf]

@AlixBa

- Drain futures will complete when there are no more messages or the timeout expires.
- connection drain is used to drain subscriptions that use the default dispatcher, so in your example there are none.
- connection drain will close the connection, so be sure you want to do this.
- In your example there are no more messages pretty much by the time you get to the line of code to drain the dispatcher, so I would expect the future to complete successfully

Try this code:

public static void main(String[] args) {
    Options options = Options.builder()
        .server("nats://localhost:4222")
        .errorListener(new ErrorListener() {})
        .build();

    try (Connection connection = Nats.connect(options)) {
        AtomicInteger published = new AtomicInteger(0);
        AtomicInteger handled = new AtomicInteger(0);

        Dispatcher d = connection.createDispatcher();
        Subscription subscription = d.subscribe(
            "subject",
            m -> {
                System.out.println("[" + System.currentTimeMillis() + "] Handler: " + new String(m.getData()));
                handled.incrementAndGet();
            }
        );

        Thread t = new Thread(() -> {
            boolean done = false;
            int x = 0;
            while (!done) {
                try {
                    connection.publish("subject", (++x + "").getBytes());
                    published.incrementAndGet();
                    Thread.sleep(10);
                }
                catch (IllegalStateException ise) {
                    System.out.println("[" + System.currentTimeMillis() + "] Publish: " + ise.getMessage());
                    done = true;
                }
                catch (InterruptedException e) {
                    System.out.println("[" + System.currentTimeMillis() + "] " + e.getMessage());
                    done = true;
                    Thread.currentThread().interrupt();
                }
            }
        });
        t.start();

        Thread.sleep(500);

        System.out.println("[" + System.currentTimeMillis() + "] About to drain dispatcher...");
        CompletableFuture<Boolean> dDrained = subscription.drain(Duration.ofSeconds(10));
        System.out.println("[" + System.currentTimeMillis() + "] Dispatcher drained: " + dDrained.get());

        System.out.println("[" + System.currentTimeMillis() + "] Published Total: " + published.get());
        System.out.println("[" + System.currentTimeMillis() + "] Handled Total: " + handled.get());
    }
    catch (Exception e) {
        System.out.println("[" + System.currentTimeMillis() + "] Exception: " + e);
    }
}

[AlixBa]

connection drain is used to drain subscriptions that use the default dispatcher, so in your example there are none.

Ha! Indeed, I didn't know this. Is this a technical limitation? From the API, I would expect it to work the same with or without the default handler. This is how I expected my code to behave

1. publish 5 messages
2. the connection receives the message 
3. the messages are pushed to the subscription's queue
4. the drain starts (using subscription.drain)
5. the subscription processes all buffered messages (then return CompletableFuture.completed(true))
6. the subscription is drained / end

I tested with

    val dispatcher = connection.createDispatcher { message =>
      println("received message " + String(message.getData()))
      Thread.sleep(1000)
    }
    val subscription = dispatcher.subscribe("subject")
    // ...
    subscription.drain(Duration.ofSeconds(30)).get()

and indeed, this times it waits properly until the 5 messages are processed.
But in this case the dispatcher also becomes inactive after the drain. I guess I was expecting subscription and dispatcher to be more independent.

[scottf]

@AlixBa Moved to a discussion.

I probably should have noted the similarities between the connection drain and dispatcher drain. They are both done in a way that expects the user to be meaning "I'm completely done with this", i.e. and they both basically close.

What were you trying to accomplish by using drain? Maybe that's a better way to look at this.

If you are done with the dispatcher, you can just unsubscribe, the dispatcher is basically just doing that, but also the closing down.

[AlixBa]

First, I can solve my issues by creating one dispatcher for one subscription, so I am not blocked at any level.

I'll explain my understanding on the dispatcher/subscription design based on the API and a little bit of code digging (even though I realize that it's not working the way I thought haha). I'll omit the synchronous subscriptions created from the connection for the next explanations

From the API POV, I can see 3 main resources (as in open/close)

• connection Nats.connect / connection.drain / connection.close
• dispatcher connection.createDispatcher / dispatcher.drain / connection.closeDispatcher
• subscription dispatcher.subscribe / dispatcher.unsubscribe

From this I kind of expect the following parent-children relationship

• connection 1-* dispatchers
• dispatchers 1-* subscriptions

Which lead me to think that

• an app should chose to use 1 or N dispatcher depending on how many callback thread they need (to simulate some kind of sequencing or parallelism)
• a subscription is a "unit of work", it's independent, owned by a dispatcher but kind of live its life. In a multi-subscriptions dispatcher, I would expect the callback of the dispatcher to call synchronously the callback of the subscription before going to the next message.
• when you drain one resource, you drain the children resources and itself, but you don't have any impact on the "parent" (counter example in NATS, draining the subscription also closes the dispatcher. I haven't checked if it's the case on a multi subscription dispatcher, but on a single one, it does).

I thought that

Dispatcher d = conn.createDispatcher();
Subscription s1 = d.subscribe("sub", msg -> {});
Subscription s2 = d.subscribe("sub2", msg -> {});
// ...
s1.drain(); 
// ...

this drain would drain only the subscription s1 (isolated from any draining on s2 or d), by processing all messages buffered for s1 until we reach the time of the drain trigger. Then, at that point, d and s2 should still be running. note(bis) I haven't tested this, only on a single subscription+dispatcher setup. But from my tests, as the issue states as soon as I called drained, the buffered messages were dropped, the subscription was close as well as the dispatcher. Even on this setup I would expect the dispatcher to still be alive as I didn't call any drain or close on it.

And I confirm that connection.drain and dispatcher.drain both work as expected. And maybe that's where I get confused. The subscription.drain doesn't seem to have the same semantic (but it's heavily based on my representation of a subscription)