Notification when Durable consumer is cleaned up

[tilenkocbek]

Hello!
We have an application where we create multiple durable consumers with specific "InactiveThreshold" property set for each consumer. Is there any possibility to be notified when consumer is deleted/cleaned-up?

The only way that I found was to call GetConsumerInformation() on a subscription and if this returns null, it means that consumer is removed. Although this works, I wonder if there is any other way to get notified, e.g. some event maybe?

Thanks!

[scottf]

You can use the management functions GetConsumerInfo/GetConsumerNames , but GetConsumerInfo is what is called from GetConsumerInformation(). Either way it's a round trip to the server.

There may be server advisories that you could subscribe to (advisories are subjects the server publishes too) but you would have to ask that in the general slack, I'm not sure if they are documented.

Are you having issues with the feature? Could you explain your use case.

[scottf]

Hmm Maybe not. I'm going to have to think about this, heartbeats will only help when there is a disconnection, but there may be some way under the new simplification api. Simplification api is an easier way to consume messages using pull consumers.

[tilenkocbek]

Thanks for your replies!
I have another question, but this time regarding subscriptions and reading messages. We have a case where we want to read all messages from the start to the end of a stream.

In the examples I see that catching NATSTimeoutException is the way to know that there are no more messages to read from the stream, e.g.

            bool keepListening = true;
            while (keepListening)
            {
                try
                {
                    Msg msg = subscription.NextMessage(50);
                    msg.Ack();
                }
                catch (NATSTimeoutException)
                {
                    keepListening = false;
                }
            }

But could we instead use Subscription.PendingMessages property to read until there are no more messages to read?
e.g.:

            while(subscription.PendingMessages > 0)
            {
                Msg msg = subscription.NextMessage();
                msg.Ack();
            }

Is there any downside of using subscription.PendingMessages instead of catching the NATSTimeoutException?

Thanks!

[scottf]

The PendingMessages will help but also you may have to consider how you messages are being published. Let's say that the stream has 100 messages. As long as no messages are added to the stream and all messages are acked within the ack wait timeout, pending will work fine. But if messages are coming in at the same time you are reading, pending will go up and down and won't be representative of what you have left.
Maybe you can set your timeout shorter but loop some more?

[scottf]

See the next comment for some sample code to demonstrate.

[scottf]

public static void Main(string[] args)
{
    try
    {
        Options opts = ConnectionFactory.GetDefaultOptions();
        opts.Url = "nats://localhost:4222";

        using (IConnection c = new ConnectionFactory().CreateConnection(opts))
        {
            IJetStreamManagement jsm = c.CreateJetStreamManagementContext();

            StreamConfiguration sc = StreamConfiguration.Builder()
                .WithName("stream")
                .WithStorageType(StorageType.Memory)
                .WithSubjects("subject")
                .Build();

            jsm.AddStream(sc);

            IJetStream js = c.CreateJetStreamContext();
            
            jsm.AddOrUpdateConsumer("stream", ConsumerConfiguration.Builder()
                .WithDurable("durable")
                .WithDeliverSubject("deliversubject")
                .WithAckPolicy(AckPolicy.None)
                .Build());


            for (int x = 1; x <= 10; x++) {
                js.Publish("subject", Encoding.UTF8.GetBytes("data" + x));
            }

            PushSubscribeOptions pso = PushSubscribeOptions.BindTo("stream", "durable");
            IJetStreamPushSyncSubscription sub = js.PushSubscribeSync(null, pso);
            for (int x = 1; x <= 5; x++) {
                Console.WriteLine(sub.NextMessage(200).MetaData);
            }

            for (int x = 11; x <= 20; x++) {
                js.Publish("subject", Encoding.UTF8.GetBytes("data" + x));
            }

            for (int x = 6; x <= 20; x++) {
                Console.WriteLine(sub.NextMessage(200).MetaData);
            }
            
            Console.WriteLine("Should Timeout");
            sub.NextMessage(200);
        }
    }
    catch (Exception ex)
    {
        Console.Error.WriteLine(ex);
    }
}

[tilenkocbek]

Great, thanks for all of your answers @scottf !