Function TestJetStream_Consumer

jetstream/test/jetstream_test.go:1654–1768

(t *testing.T)

}

func TestJetStream_Consumer(t *testing.T) {
	tests := []struct {
		name      string
		stream    string
		durable   string
		timeout   time.Duration
		withError error
	}{
		{
			name:    "get existing consumer",
			stream:  "foo",
			durable: "dur",
			timeout: 10 * time.Second,
		},
		{
			name:    "with empty context",
			stream:  "foo",
			durable: "dur",
		},
		{
			name:      "consumer does not exist",
			stream:    "foo",
			durable:   "abc",
			timeout:   10 * time.Second,
			withError: jetstream.ErrConsumerNotFound,
		},
		{
			name:      "invalid durable name",
			stream:    "foo",
			durable:   "dur.123",
			withError: jetstream.ErrInvalidConsumerName,
		},
		{
			name:      "stream does not exist",
			stream:    "abc",
			durable:   "dur",
			withError: jetstream.ErrStreamNotFound,
		},
		{
			name:      "invalid stream name",
			stream:    "foo.1",
			durable:   "dur",
			withError: jetstream.ErrInvalidStreamName,
		},
		{
			name:      "context timeout",
			stream:    "foo",
			durable:   "dur",
			timeout:   1 * time.Microsecond,
			withError: context.DeadlineExceeded,
		},
	}

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