.Net Nats Client Reconnection

[alberk8]

I would like to know how to re start the ConsumeAsync when the connection to server is re-established?. At the moment I can see that the connection is close and re-open but the ConsumeAsync does not quit or re-start the consume for message that is pending when the connection was dropped/closed (server stopped or not contactable due to network issue).

using NATS.Client.Core;
using NATS.Client.JetStream;
using NATS.Client.JetStream.Models;
using NATS.Client.Serializers.Json;


namespace ConsoleNatsSubscriber
{
    internal class Program
    {
        static async Task Main(string[] args)
        {
           
            Console.WriteLine("Hello, World!");

            var opt = new NatsOpts
            {
                Url = "nats://127.0.0.1:4222",
                SerializerRegistry = NatsJsonSerializerRegistry.Default,
                Name = "Nats_Connect",
                  
            };

            var connect = new NatsConnection(opt);
            connect.ConnectionDisconnected += Connect_ConnectionDisconnected;
            connect.ConnectionOpened += Connect_ConnectionOpened;
            connect.ReconnectFailed += Connect_ReconnectFailed;

            var jsContext = new NatsJSContext(connect);

            var stream = await jsContext.CreateStreamAsync(new StreamConfig("EVENTS", new[] { "data.>" }));

            var consumerName = "processor";
            var ackWait = TimeSpan.FromSeconds(10);
            var ackPolicy = ConsumerConfigAckPolicy.Explicit;

            var consumer = await jsContext.CreateOrUpdateConsumerAsync("EVENTS", new ConsumerConfig(consumerName)
            {
                DeliverPolicy = ConsumerConfigDeliverPolicy.All,
                FilterSubject = "data.>",
                AckPolicy = ackPolicy,
                AckWait = ackWait,
 
            });
     

       
            
            await foreach (NatsJSMsg<Person> msg in consumer.ConsumeAsync<Person>( ))
            {
                Console.WriteLine($"Subject: {msg.Subject} Size: {msg.Size}");
                Console.WriteLine($"Name: {msg.Data.Name} Address: {msg.Data.Address}");
               
                var md = msg.Metadata;
                if (md.HasValue)
                {

                    Console.WriteLine($"Num Delivered: {md.Value.NumDelivered} TimeStamp: {md.Value.Timestamp.ToLocalTime()}");
                    Console.WriteLine($"Consumer Seq: {md.Value.Sequence.Consumer} Stream Seq: {md.Value.Sequence.Stream}");
                }
                
            }
            
            Console.WriteLine("Done");
            Console.ReadKey();
        }

        private static ValueTask Connect_ReconnectFailed(object? sender, NatsEventArgs args)
        {
           Console.WriteLine($"Connect failed: {args.Message}");
          return ValueTask.CompletedTask;
        }

        private static ValueTask Connect_ConnectionOpened(object? sender, NatsEventArgs args)
        {
            Console.WriteLine($"Connect Open: {args.Message}");
             return ValueTask.CompletedTask;
        }

        private static ValueTask Connect_ConnectionDisconnected(object? sender, NatsEventArgs args)
        {
            Console.WriteLine($"Connect Disconnect: {args.Message}");
            return ValueTask.CompletedTask;
        }

        public class Person
        {
            public string Name { get; set; } = string.Empty;
            public string Address { get; set; } = string.Empty;
        }
    }
}

[mtmk]

it should recover automatically in most cases, so you don't have to do anything.

But, if you see a case where we can't recover, you can catch the exceptions and retry with a back off. Some documentation here: https://nats-io.github.io/nats.net.v2/documentation/jetstream/consume.html#handling-exceptions

Example:

while (!cancellationToken.IsCancellationRequested)
{
    try
    {
        await consumer.RefreshAsync(cancellationToken); // or try to recreate consumer

        await foreach (var msg in consumer.ConsumeAsync<Order>(serializer: orderSerializer).WithCancellation(cancellationToken))
        {
            // Process message
            await msg.AckAsync(cancellationToken: cancellationToken);
        }
    }
    catch (NatsJSProtocolException e)
    {
        // log exception
    }
    catch (NatsJSException e)
    {
        // log exception
        await Task.Delay(Random.Shared.Next(1000)); // jitter
        await Task.Delay(backoffSeries(retry++), cancellationToken); // backoff
    }
}

[mtmk]

yes but how are you getting there? could you post a Program.cs with the steps to reproduce?

[alberk8]

using NATS.Client.Core;
using NATS.Client.JetStream;
using NATS.Client.JetStream.Models;
using NATS.Client.Serializers.Json;
using System.Text;
using System.Threading;

namespace ConsoleNatsSubscriber
{
    internal class Program
    {
        private static CancellationTokenSource cancellation = new CancellationTokenSource();
        static async Task Main(string[] args)
        {
           
            Console.WriteLine("Hello, World!");

            var opt = new NatsOpts
            {
                Url = "nats://127.0.0.1:4222",
                SerializerRegistry = NatsJsonSerializerRegistry.Default,
                Name = "Nats_Connect",
                  
            };

            var connect = new NatsConnection(opt);
            connect.ConnectionDisconnected += Connect_ConnectionDisconnected;
            connect.ConnectionOpened += Connect_ConnectionOpened;
            connect.ReconnectFailed += Connect_ReconnectFailed;

            var jsContext = new NatsJSContext(connect);
            
  
            var stream = await jsContext.CreateStreamAsync(new StreamConfig("EVENTS", new[] { "data.>" }));
       
            var consumerName = "processor";
            var ackWait = TimeSpan.FromSeconds(10);
            var ackPolicy = ConsumerConfigAckPolicy.Explicit;
     

           

            var consumer = await jsContext.CreateOrUpdateConsumerAsync("EVENTS", new ConsumerConfig(consumerName)
            {
                DeliverPolicy = ConsumerConfigDeliverPolicy.All,
                FilterSubject = "data.>",
                AckPolicy = ackPolicy,
                AckWait = ackWait,
                 // Only for Push
                 //IdleHeartbeat = TimeSpan.FromSeconds(3),
                MaxDeliver = 5,
                Backoff = new List<long>()
                {
                            (long) TimeSpan.FromSeconds(3).TotalMicroseconds * 1_000L,
                            (long) TimeSpan.FromSeconds(5).TotalMicroseconds * 1_000L,
                            //(long) TimeSpan.FromSeconds(10).TotalMicroseconds * 1_000L,
                            //(long) TimeSpan.FromSeconds(15).TotalMicroseconds * 1_000L
                }

            });


            
            while (true) //!cancellationToken.IsCancellationRequested)
            {
                Console.WriteLine("While Loop");
                var cancellationToken = cancellation.Token;
                try
                {
                    await stream.RefreshAsync(cancellationToken);
                    await consumer.RefreshAsync(cancellationToken);
                   
		    await foreach (NatsJSMsg<Person> msg in consumer.ConsumeAsync<Person>().WithCancellation(cancellationToken))
                    {
                        Console.WriteLine($"Subject: {msg.Subject} Size: {msg.Size}");
                        Console.WriteLine($"Name: {msg.Data.Name} Address: {msg.Data.Address}");
                       

                        var md = msg.Metadata;
                        if (md.HasValue)
                        {

                            Console.WriteLine($"Num Delivered: {md.Value.NumDelivered} TimeStamp: {md.Value.Timestamp.ToLocalTime()}");
                            Console.WriteLine($"Consumer Seq: {md.Value.Sequence.Consumer} Stream Seq: {md.Value.Sequence.Stream}");
                        }
                        //await msg.NakAsync(delay: TimeSpan.FromSeconds(1));
                    }
                }
                catch (NatsJSProtocolException e)
                {
                    // log exception
                    Console.WriteLine("NSJS Protocol Exception: " + e.Message);
                }
                catch (NatsJSException e)
                {
                    Console.WriteLine("NSJS Exception: " + e.Message);
                    // log exception
                    await Task.Delay(Random.Shared.Next(1000)); // jitter

                    //await Task.Delay(backoffSeries(retry++), cancellationToken); // backoff
                }
                catch(Exception ex)
                {
                    Console.WriteLine("Other Exceptions: " + ex.Message);
                    
                }
                await Task.Delay(1000);
            }

       
            Console.WriteLine("Done");
            Console.ReadKey();
        }

        private static ValueTask Connect_ReconnectFailed(object? sender, NatsEventArgs args)
        {
           Console.WriteLine($"Connect failed: {args.Message}");
          return ValueTask.CompletedTask;
        }

        private static ValueTask Connect_ConnectionOpened(object? sender, NatsEventArgs args)
        {
            Console.WriteLine($"Connect Open: {args.Message}");
            cancellation = new CancellationTokenSource();
          
            return ValueTask.CompletedTask;
        }

        private static ValueTask Connect_ConnectionDisconnected(object? sender, NatsEventArgs args)
        {
            Console.WriteLine($"Connect Disconnect: {args.Message}");
            cancellation.Cancel();
            return ValueTask.CompletedTask;
        }

        public class Person
        {
            public string Name { get; set; } = string.Empty;
            public string Address { get; set; } = string.Empty;
        }
    }
}

[mtmk]

this is working for me. what are you doing to get that error? are you deleting the stream?

[alberk8]

So sorry, I was running the NATS docker container with --rm switch. Stop and restarting the container essentially start with a blank slate. The code above works when the docker is started without the remove switch. Thank you again for your patience.

[mtmk]

no worries :) glad it worked 💯