Function TestStreamResetConsumer

jetstream/test/stream_test.go:1747–1827

(t *testing.T)

}

func TestStreamResetConsumer(t *testing.T) {
	srv := RunBasicJetStreamServer()
	defer shutdownJSServerAndRemoveStorage(t, srv)

	nc, err := nats.Connect(srv.ClientURL())
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}
	defer nc.Close()

	js, err := jetstream.New(nc)
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}

	ctx := context.Background()
	s, err := js.CreateStream(ctx, jetstream.StreamConfig{
		Name:     "TEST",
		Subjects: []string{"foo"},
	})
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}

	for i := range 10 {
		if _, err := js.Publish(ctx, "foo", fmt.Appendf(nil, "msg-%d", i)); err != nil {
			t.Fatalf("Unexpected error: %v", err)
		}
	}

	cons, err := s.CreateOrUpdateConsumer(ctx, jetstream.ConsumerConfig{
		Durable:   "cons",
		AckPolicy: jetstream.AckExplicitPolicy,
	})
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}

	// Drain 5 messages without ack so ack_floor stays 0.
	batch, err := cons.Fetch(5)
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}
	count := 0
	for range batch.Messages() {
		count++
	}
	if count != 5 {
		t.Fatalf("Expected 5 messages, got %d", count)
	}

	resp, err := s.ResetConsumer(ctx, "cons")
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}
	if resp.ResetSeq != 1 {
		t.Fatalf("Expected ResetSeq=1, got %d", resp.ResetSeq)
	}