How to create non-blocking consumer and call delegate on new message received

[rdkumavat1]

How to create a non-blocking consumer subscription and call a method with a delegate on new message publish.

[mtmk]

// dotnet add package NATS.Net
// dotnet add package Microsoft.Extensions.Logging.Console

var opts = new NatsOpts
{
    LoggerFactory = LoggerFactory.Create(builder => builder.AddConsole()),
};

await using var nats = new NatsConnection(opts);
var js = new NatsJSContext(nats);

var stream = await js.CreateStreamAsync(new StreamConfig("MY_STREAM", ["events.>"]));
var consumer = await stream.CreateOrUpdateConsumerAsync(new ConsumerConfig("MY_CONSUMER"));

var cts = new CancellationTokenSource();

async Task MessageDelegate(NatsJSMsg<string> msg)
{
    Console.WriteLine($"processing: {msg}");
    await msg.AckAsync(cancellationToken: cts.Token);
}

var consumeTask = Task.Run(async () =>
{
    await foreach (var msg in consumer.ConsumeAsync<string>(cancellationToken: cts.Token))
    {
        await MessageDelegate(msg);
    }
});

// Stop application
await cts.CancelAsync();
await consumeTask;

[rdkumavat1]

Thank you, I am also looking for NATS to notify the consumer on new message arrival, without subscriber has to actively pool, If I have more than one subscribers then I will end-up in creating pooling for all of those, I want to avoid that

[mtmk]

if you can give me a code sample I can try to help. sorry not 100% sure what you're trying to do.

[rickdotnet]

@rdkumavat1 , you first need to figure out what subjects you're wanting to listen on. Depending on your use of wildcards, you could get a way with one subscription. Then, you can fan the messages out in any way you need. If you require multiple subscriptions, you use something like a Channel to determine the flow of messages from there.