Function TestJetStream_CreateConsumer

jetstream/test/jetstream_test.go:1425–1530

(t *testing.T)

}

func TestJetStream_CreateConsumer(t *testing.T) {
	tests := []struct {
		name           string
		consumerConfig jetstream.ConsumerConfig
		shouldCreate   bool
		stream         string
		withError      error
	}{
		{
			name:           "create consumer",
			consumerConfig: jetstream.ConsumerConfig{Durable: "dur"},
			stream:         "foo",
			shouldCreate:   true,
		},
		{
			name:           "consumer already exists, error",
			consumerConfig: jetstream.ConsumerConfig{Durable: "dur", Description: "test consumer"},
			stream:         "foo",
			withError:      jetstream.ErrConsumerExists,
		},
		{
			name:           "stream does not exist",
			stream:         "abc",
			consumerConfig: jetstream.ConsumerConfig{Durable: "dur"},
			withError:      jetstream.ErrStreamNotFound,
		},
		{
			name:      "invalid stream name",
			stream:    "foo.1",
			withError: jetstream.ErrInvalidStreamName,
		},
	}

	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)
		}

		for _, test := range tests {
			t.Run(fmt.Sprintf("%s %s", consType, test.name), func(t *testing.T) {
				var sub *nats.Subscription
				if test.consumerConfig.FilterSubject != "" {