Function TestAckVariants

jetstream/test/message_test.go:89–405

(t *testing.T)

}

func TestAckVariants(t *testing.T) {
	setup := func(ctx context.Context, t *testing.T) (*server.Server, *nats.Conn, jetstream.JetStream, jetstream.Consumer) {
		srv := RunBasicJetStreamServer()

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

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

		s, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "foo", Subjects: []string{"FOO.*"}})
		if err != nil {
			t.Fatalf("Unexpected error: %v", err)
		}
		c, err := s.CreateOrUpdateConsumer(ctx, jetstream.ConsumerConfig{
			Durable:     "cons",
			AckPolicy:   jetstream.AckExplicitPolicy,
			Description: "test consumer",
		})
		if err != nil {
			t.Fatalf("Unexpected error: %v", err)
		}
		return srv, nc, js, c
	}

	t.Run("standard ack", func(t *testing.T) {
		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		srv, nc, js, c := setup(ctx, t)
		defer shutdownJSServerAndRemoveStorage(t, srv)
		defer nc.Close()

		if _, err := js.Publish(ctx, "FOO.1", []byte("msg")); err != nil {
			t.Fatalf("Unexpected error: %v", err)
		}
		msgs, err := c.Fetch(1)
		if err != nil {
			t.Fatalf("Unexpected error: %v", err)
		}
		msg := <-msgs.Messages()
		if msg == nil {
			t.Fatalf("No messages available")
		}
		if err := msgs.Error(); err != nil {
			t.Fatalf("unexpected error during fetch: %v", err)
		}
		sub, err := nc.SubscribeSync(msg.Reply())
		if err != nil {
			t.Fatalf("Unexpected error: %v", err)
		}

		if err := msg.Ack(); err != nil {
			t.Fatalf("Unexpected error: %v", err)
		}