PullSubscribe Fetch stops working after losing connection to server

[bhosale-ajay]

Following code uses PullSubscribe and read 1 message at a time. It works well, as long as connection is maintained. If connection to server is lost then even after reestablishing the subscription is unable to fetch any messages.

The only option worked for me is using the DisconnectedEventHandler and reset the connection and subscription. Documentation mentions that client library will automatically re-establish all the subscriptions, there is nothing for the application programmer to do.

Any pointers will be helpful.

NATS Library Version : 1.1.0

// Want to establish connection only in the infinite loop for resiliency
IConnection connection = null;
IJetStreamManagement jsm = null;
ConsumerConfiguration cc = ConsumerConfiguration.Builder()
    .WithDurable(queueName)
    .WithFilterSubject(subject)
    .WithAckWait(AckWaitInMillSeconds)
    .Build();
PullSubscribeOptions pullOptions = null;
IJetStreamPullSubscription subscription = null;
NATS.Client.Options opts = ConnectionFactory.GetDefaultOptions(SERVER);
opts.AllowReconnect = true;
// Temp Fix - Part 1
// bool disconnected = false;
// opts.DisconnectedEventHandler += (sender, args) =>
// {
//     Console.WriteLine("Disconnected.");
//     disconnected = true;
// };
while (!ct.IsCancellationRequested)
{
    try
    {
        connection ??= new ConnectionFactory().CreateConnection(opts);
        jsm ??= connection.CreateJetStreamManagementContext();
        if (pullOptions == null)
        {
            jsm.AddOrUpdateConsumer(streamName, cc);
            pullOptions = PullSubscribeOptions.BindTo(streamName, queueName);
        }
        if (connection.IsClosed() ||
            connection.IsDraining() ||
            connection.IsReconnecting())
        {
            ct.WaitHandle.WaitOne(WaitingPeriodWhenNoConnectionIsAvailable);
            continue;
        }
        Console.WriteLine("Pulling message ...");
        subscription ??= connection.CreateJetStreamContext().PullSubscribe(subject, pullOptions);
        message = subscription.Fetch(1, WaitingPeriodForNextMessageInMillSeconds).FirstOrDefault();
        // Process Message
    }
    catch (NATSTimeoutException)
    {
        logger.LogError("Nats timeout exception {consumer} for {subject}.", queueName, subject);
        ct.WaitHandle.WaitOne(WaitingPeriodWhenNoMessagesAreAvailable);
    }
    catch (NATSJetStreamClientException ex)
    {
        logger.LogError(ex, "Error while setting up {consumer} for {subject}.", queueName, subject);
        ct.WaitHandle.WaitOne(WaitingPeriodWhenNoConnectionIsAvailable);
    }
    catch (Exception ex)
    {
        logger.LogError(ex, "Error while fetching message for {consumer} {subject}.", queueName, subject);
        ct.WaitHandle.WaitOne(WaitingPeriodWhenNoConnectionIsAvailable);
    }
    ct.WaitHandle.WaitOne(SleepPeriod);
    // Temp fix - Part 2
    // if (disconnected)
    // {
    //     if (subscription != null)
    //     {
    //         subscription.Dispose();
    //         subscription = null;
    //     }
    //     if (connection != null)
    //     {
    //         connection.Close();
    //         connection.Dispose();
    //         jsm = null;
    //         connection = null;
    //     }
    //     disconnected = false;
    // }
}

[scottf]

Can you upgrade to a more recent version and try again? There have been improvements for this specific issue. Consumer must be durable or at least have an inactive threshold long enough to survive the outage.