Is it possible to create an ephemeral consumer that can fetch only the most recent messages in the stream?

[aliasadidev]

Is it possible to fetch only the latest 3 new messages from the stream? Currently, using DeliverPolicy.All will send messages starting from sequence number 1. It would be great to have an ephemeral consumer that can display only the latest messages based on a count.

import { AckPolicy, connect, DeliverPolicy, StringCodec } from "nats";

const server = { servers: "nats://localhost:8222" };
const nc = await connect(server);
const jsm = await nc.jetstreamManager();
const streamName = "alarm-stream";
const js = await jsm.streams.get(streamName);
const sub = "alarm-stream-subjects.*";
const consumerName = "alarm-stream-consumer-1234";

jsm.consumers.add(streamName, {
  name: consumerName,
  ack_policy: AckPolicy.None,
  filter_subject: sub,
  deliver_policy: DeliverPolicy.All, // ** Can we have DeliverPolicy.AllFromLast **
});

const c = await js.getConsumer(consumerName);
const sc = StringCodec();

var result = await c.fetch({ max_messages: 3 }); // Consume last 3 messages
for await (const m of result) {
  console.log("seq:", m.seq, sc.decode(m.data));
}

await nc.close();

[Jarema]

Hey!

No, you can't create a consumer that "knows" to fetch just the last n messages.

However what you can do is run stream info, check the last sequence, and create an ordered consumer starting fom that sequence to get those few messages.

In next release (2.11) there will also be a way to get multiple messages from stream without creating consumer with Direct Batch Get. However you would still need to check the sequences on the stream first.

[derekcollison]

If you just want new messages since time of creation of the consumer you can use DeliverNew().

[aliasadidev]

I've previously used this option, but it's ineffective here since some messages already exist within the stream.

[derekcollison]

ok yes in re-reading you want last 3 plus all new after. So @Jarema is correct, you need to get StreamInfo and calculate a starting sequence number to create the consumer.

[aliasadidev]

I attempted to use the ordered consumer, but some messages have been acknowledged from the stream, and the sequence number is not valid for subtraction from the current sequence number.

[aliasadidev]

in this case I'm losing ack messages

            stream = await js.GetStreamAsync("alarm-jetstream");

            var consumer = await stream.CreateOrderedConsumerAsync(new NatsJSOrderedConsumerOpts
            {
                DeliverPolicy = ConsumerConfigDeliverPolicy.ByStartSequence,
                OptStartSeq = (ulong)stream.Info.State.LastSeq - 3,
            });

            await foreach (var msg in consumer.FetchAsync<AlarmMessage>(opts: new NatsJSFetchOpts()
            {
                MaxMsgs = 3
            }, cancellationToken: cancellationToken))
            {
                System.Console.WriteLine(msg.Data);
            }

[Jarema]

Ack does not matter for the stream, unless it's a workquęue / interest stream.
Ack is a consumer construct that allows it to track its view into the stream.
Stream is independent.

I'm not sure what you mean by sequence not being valid.
Can you elaborate?

Also yes - Ordered Consumer does not have acks at all. It's client side code that tracks sequences and recreates consumer whenever it finds a gap in stream sequence. You can as well use a standard consumer if you need acks for whatever reason.