Is it possible to create a consumer that receives all available messages at the time of creation and then automatically unsubscribes?

[hayleyjames]

Hi,

I'm trying to use JetStream as an event store and I'd like to create a consumer that knows when it has seen all the messages on a stream.

I have achieved that behaviour using a pull subscriber, but using a timeout doesn't seem like a reliable way to know when all messages have been processed. Is there a better way that I'm missing?

sub, err := js.PullSubscribe("", "", nats.BindStream("CONFIGS"), nats.AckNone(), nats.DeliverAll())
if err != nil {
	panic(err)
}
defer sub.Unsubscribe()

for {
	ctx, cancel := context.WithTimeout(context.Background(), 250*time.Millisecond)
	defer cancel()

	msgs, err := sub.Fetch(10, nats.Context(ctx))
	if err != nil {
		if errors.Is(err, context.DeadlineExceeded) {
			break
		}

		panic(err)
	}

	for _, msg := range msgs {
		println(msg.Subject, string(msg.Data))
	}
}
println("Fetched all messages")

Thanks,
Hayley

[bruth]

@hayleyjames There is no consumer option for that today, but what you can do is do a stream info lookup to get the current sequence (at the time of the lookup) and then consume until you have reached that sequence number to cleanly stop.

[derekcollison]

You have two options, each involve looking at the metadata associated with the delivered message. You could exit when num pending goes to 0, but if active stream that could mean messages after you created the consumer. You could also get stream info before you start, mark last stream sequence, and unsubscribe from the consumer when you see that sequence. Same trick could be applicable to time but there could be a drift between local clock where consumer is created and the timestamp on the delivered message.

[tpihl]

There is a similar solution that may be easier thanks to @piotrpio and @Jarema

create the consumer
remember the number of msgs you will receive right away;
remaining:= con.CachedInfo().NumPending
now consume the msgs anyway you like, like with a fetch for them all, for ;remaining>0;remaining--{ or however
now remove the consumer and you're done.

Seems like the CachedInfo() contains correct info while not requiring an extra roundtrip.