Function TestListConsumers

jetstream/test/stream_test.go:1325–1402

(t *testing.T)

}

func TestListConsumers(t *testing.T) {
	tests := []struct {
		name         string
		consumersNum int
		timeout      time.Duration
		withError    error
	}{
		{
			name:         "list consumers",
			consumersNum: 500,
			timeout:      5 * time.Second,
		},
		{
			name:         "with empty context",
			consumersNum: 500,
		},
		{
			name:         "no consumers available",
			consumersNum: 0,
		},
		{
			name:         "context timeout",
			consumersNum: 500,
			timeout:      1 * time.Microsecond,
			withError:    context.DeadlineExceeded,
		},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			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()
			s, err := js.CreateStream(context.Background(), jetstream.StreamConfig{Name: "foo", Subjects: []string{"FOO.*"}})
			if err != nil {
				t.Fatalf("Unexpected error: %v", err)
			}
			for i := 0; i < test.consumersNum; i++ {
				_, err = s.CreateOrUpdateConsumer(context.Background(), jetstream.ConsumerConfig{AckPolicy: jetstream.AckExplicitPolicy})
				if err != nil {
					t.Fatalf("Unexpected error: %v", err)
				}
			}
			ctx := context.Background()
			if test.timeout > 0 {
				var cancel context.CancelFunc
				ctx, cancel = context.WithTimeout(ctx, test.timeout)
				defer cancel()
			}