Function TestConsumer

jetstream/test/stream_test.go:429–517

(t *testing.T)

}

func TestConsumer(t *testing.T) {
	tests := []struct {
		name      string
		durable   string
		withError error
	}{
		{
			name:    "get existing consumer",
			durable: "dur",
		},
		{
			name:      "consumer does not exist",
			durable:   "abc",
			withError: jetstream.ErrConsumerNotFound,
		},
		{
			name:      "invalid durable name",
			durable:   "dur.123",
			withError: jetstream.ErrInvalidConsumerName,
		},
		{
			name:      "empty durable name",
			durable:   "",
			withError: jetstream.ErrInvalidConsumerName,
		},
		{
			name:      "empty consumer name",
			durable:   "",
			withError: jetstream.ErrInvalidConsumerName,
		},
	}

	srv := RunBasicJetStreamServer()
	defer shutdownJSServerAndRemoveStorage(t, srv)
	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)
	}
	defer nc.Close()

	for _, consType := range []string{"pull", "push"} {
		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		s, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "foo", Subjects: []string{"FOO.*"}})
		if err != nil {
			t.Fatalf("Unexpected error: %v", err)
		}
		if consType == "pull" {
			_, err = s.CreateOrUpdateConsumer(ctx, jetstream.ConsumerConfig{Durable: "dur", AckPolicy: jetstream.AckAllPolicy, Description: "desc"})
		} else {
			_, err = s.CreateOrUpdatePushConsumer(ctx, jetstream.ConsumerConfig{Durable: "dur", DeliverSubject: "inbox", AckPolicy: jetstream.AckAllPolicy, Description: "desc"})
		}
		if err != nil {