[QUESTION] - How to restrict consuming previously consumed messages?

[sujit-baniya]

I'm trying to learn and implement message broker in my app using nats.io and jetstream but got some weird issue.

The repo I'm trying with: https://github.com/oarkflow/nats-app

To start server: go run server.go

In a test, I published some messages using go run publisher.go while consumer is not UP. Now whenever I try to run consumer go run consumer.go it re-executes all messages previously published. I'm not sure what mistake am I doing.

Please suggest any solution to the problem.

[Jarema]

Consumer is a server side construct, so it definately can track the progress your app is making, no matter if you restart it or not.
However, every call to Subscribe in a fashion you are doing it creates a new one, hence the result you see.

You need to create the Consumer first, then start fetching messagess. That is a much easier way to get what is happening.

Here is a good example with the new API:
https://github.com/nats-io/nats.go/tree/main/jetstream#continuous-polling

Also check out the video we recently released about this topic: https://youtu.be/_CN1OO7yN0I?si=R3OjCc2Qw1tcAoho

[sujit-baniya]

I created a consumer first. After it's closed/terminated and restarted again, it processed the previously consumed tasks as shown in video https://www.loom.com/share/0499a5358b4f4b8bb6f2c0961e4046d2.

[Jarema]

Yes, because you create a new consumer every single time you call that method.

Here is a full example with the new JetStream API.

    nc, _ := nats.Connect(nats.DefaultURL)
	js, _ := jetstream.New(nc)

	ctx := context.Background()
	// Get the stream. You can also create it.
	stream, err := js.Stream(ctx, "ORDERS")
	if err != nil {
		fmt.Println("Error getting stream info:", err)
	}
    // Create a consumer. If consumer with this name exists, it will be updated.
    // If it exists with the same config, it will just give you the handle to it.
    // This consumer will persist through the restarts.
	consumer, err := stream.CreateOrUpdateConsumer(ctx, jetstream.ConsumerConfig{
		Durable:       "Reviews",
		FilterSubject: "ORDERS.reviews",
	})

    // Start consuming the messages.
	consumer.Consume(func(message jetstream.Msg) {
		fmt.Println("Received message:", string(message.Data()))
		message.Ack()
	})