Difference between SubscribeAsync vs ConsumeAsync in jetstream context

[YashasAnand]

Since subscribe async is a concept of core-nats, i was under the assumption that subscribe async would not persist messages to stream when used in context of jetstream.

But i find out that SubscribeAsync is still persisitning data to stream.

Question 1) What is use case of subscribeAsync & consumeAsync

Question 2) Is there no way we can do a request-reply format in jetstream without persisiting to stream.

Requesting some assistance from community!..

[mtmk]

you can't tell publish calls to persist a message. whether a message will be persisted or not is controlled by the stream configuration where you can list a set of subjects often with wildcards.

when you subscribe you only get the new messages published from that point in time forward.

when you consume subjects set up to be persisted by a stream you can retrieve the persisted messages which might've been sent before the time of consumption.

for request reply you don't need jetstream anyway. just make sure to use subjects that are not configured by any stream to be persisted.

hope it helps

PS anyone please feel free to jump in for a different perspective and possibly a better explanation.

PPS see also https://nats-io.github.io/nats.net/documentation/jetstream/intro.html

[YashasAnand]

@mtmk Thank you for your response! I have a usecase where in my application I need few messages to be persisted using jetstream & few to not persist. I tried to subscribeAsync thinking messages wont persist on jetstream. I am using WorkQueue based retention at the moment but i plan to switch to retention with 7 days of message holding.

I observe that messages received on subscribe async was not getting remved from workQueue because i was not acking the messages (as i assumed it wont persist).

As per your above suggestion, do you suggest that I create a new subject & a subscription not connected to jetstream context? If so then can i reuse the jetstream connection object to publish & subscribe to these messges or do i need a sperate connection to core nats broker?

[mtmk]

you'd need to use different subjects e.g. orders.> could be ones you want to persist and set up as part of a ORDERS stream and events.> could be for just plain NATS core messages and not configured in any stream.

it's like layers. JetStream is built on top of NATS Core and you can observe messages being published by subscribing to the same subjects as streams configured to persist. but unless you use a consumer to get those messages from a stream they won't be taken out of a work queue stream for example.

// dotnet add package NATS.Net
using NATS.Client.JetStream;
using NATS.Client.JetStream.Models;
using NATS.Net;

var stop = new CancellationTokenSource();

await using var client = new NatsClient();

var subTask1 = Task.Run(async () =>
{
    // this won't affect the ORDERS stream at all
    // and it's not guaranteed that we see these messages (at most once delivery)
    await foreach (var msg in client.SubscribeAsync<Order>("orders.>").WithCancellation(stop.Token))
    {
        Console.WriteLine($"Seen an order being published: {msg.Subject}: {msg.Data}");
    }
});

var subTask2 = Task.Run(async () =>
{
    // again, there's no guarantee that we see these messages (at most once delivery)
    await foreach (var msg in client.SubscribeAsync<string>("events.>").WithCancellation(stop.Token))
    {
        Console.WriteLine($"Event received: {msg.Subject}: {msg.Data}");
    }
});

// Wait for the subscriptions task to be ready
await Task.Delay(1000);

var js = client.CreateJetStreamContext();

// (note this isn't a work queue stream)
await js.CreateStreamAsync(new StreamConfig("ORDERS", ["orders.>"]));

// Fire and forget message
await client.PublishAsync("events.website.visit", 1);

// Messages we want to make sure they are persisted
var ack = await js.PublishAsync("orders.new.1", new Order(1, "shoes"));
ack.EnsureSuccess();

// Another fire and forget message
await client.PublishAsync("events.shop.checkout", "success");

var consumer = await js.CreateConsumerAsync("ORDERS", new ConsumerConfig("order_processor"));
var consumeTask = Task.Run(async () =>
{
    // This will affect the order_processor consumer and the ORDERS stream if it was a WorkQueue stream
    // and it's guaranteed that we see these messages 'at least once'
    await foreach (var jsMsg in consumer.ConsumeAsync<Order>().WithCancellation(stop.Token))
    {
        Console.WriteLine($"Processing: {jsMsg.Data}");
        await jsMsg.AckAsync();

        // Stop after processing the first message for this example
        await stop.CancelAsync();
    }
});

await subTask1;
await subTask2;
await consumeTask;

record Order(int Id, string Product);

// Output:
// Event received: events.website.visit: 1
// Seen an order being published: orders.new.1: Order { Id = 1, Product = shoes }
// Event received: events.shop.checkout: success
// Processing: Order { Id = 1, Product = shoes }

// After running this a few times try these nats cli commands:
// > nats stream list
// > nats consumer report ORDERS
//
// Also try these e.g.:
// > nats sub "events.>"
// > nats pub events.website.visit 1

[YashasAnand]

@mtmk thank you so much for the your response, the key takeaway would be that using a single connection of natsClient, i am able to publish & consume to both nats jetstream related subjects & also publish & subscribe to atmost once delivery based like core nats.

Thanks again for your time!!