SubscribeAsync Process Pending message

[alberk8]

I have the code below that is a HostedService and if Do Task is taking time and I have a fast incoming message then message get dropped and or the NatsConnect gets drop. Is there a way to process every message. Maybe something to throttle the incoming message?

NatsSubOpts natsSubOpts = new() 
await foreach (var msg in _inatsConnection.SubscribeAsync<AlertMessage>("AlertMessageSubject", 
    cancellationToken: cancellationToken, opts: natsSubOpts))
{
    Console.WriteLine($"count {count++}");
    try
    {
        if (msg.Data is not null)
        {
           // Do Task 
        }
        
    }
    catch(Exception ex)
    {
        Console.WriteLine("Error: " + ex.Message);
    }

}

[mtmk]

you can disable message dropping in the options by changing the internal channel behaviour:

await using var nats = new NatsConnection(new NatsOpts
{
    SubPendingChannelFullMode = BoundedChannelFullMode.Wait,
});

There is also a SubPendingChannelCapacity option but I wouldn't recommend enlarging than the default value since the channel will start allocating on the LOH. see the discussion in #276 for more details.