Function TestJetStream_CreateOrUpdateConsumer

jetstream/test/jetstream_test.go:1290–1423

(t *testing.T)

}

func TestJetStream_CreateOrUpdateConsumer(t *testing.T) {
	tests := []struct {
		name           string
		stream         string
		consumerConfig jetstream.ConsumerConfig
		shouldCreate   bool
		timeout        time.Duration
		withError      error
	}{
		{
			name:           "create durable pull consumer",
			stream:         "foo",
			consumerConfig: jetstream.ConsumerConfig{Durable: "dur", AckPolicy: jetstream.AckExplicitPolicy},
			timeout:        10 * time.Second,
			shouldCreate:   true,
		},
		{
			name:           "create ephemeral pull consumer",
			stream:         "foo",
			consumerConfig: jetstream.ConsumerConfig{AckPolicy: jetstream.AckExplicitPolicy},
			timeout:        10 * time.Second,
			shouldCreate:   true,
		},
		{
			name:           "with empty context",
			consumerConfig: jetstream.ConsumerConfig{AckPolicy: jetstream.AckExplicitPolicy},
			stream:         "foo",
			shouldCreate:   true,
		},
		{
			name:           "consumer already exists, update",
			stream:         "foo",
			consumerConfig: jetstream.ConsumerConfig{Durable: "dur", AckPolicy: jetstream.AckExplicitPolicy, Description: "test consumer"},
		},
		{
			name:           "consumer already exists, illegal update",
			stream:         "foo",
			consumerConfig: jetstream.ConsumerConfig{Durable: "dur", AckPolicy: jetstream.AckNonePolicy, Description: "test consumer"},
			withError:      jetstream.ErrConsumerCreate,
		},
		{
			name:      "stream does not exist",
			stream:    "abc",
			withError: jetstream.ErrStreamNotFound,
		},
		{
			name:      "invalid stream name",
			stream:    "foo.1",
			withError: jetstream.ErrInvalidStreamName,
		},
		{
			name:           "invalid durable name",
			stream:         "foo",
			consumerConfig: jetstream.ConsumerConfig{Durable: "dur.123", AckPolicy: jetstream.AckExplicitPolicy},
			withError:      jetstream.ErrInvalidConsumerName,
		},
		{
			name:           "context timeout",