Function TestUpdateConsumer

jetstream/test/stream_test.go:318–427

(t *testing.T)

}

func TestUpdateConsumer(t *testing.T) {
	tests := []struct {
		name           string
		consumerConfig jetstream.ConsumerConfig
		shouldUpdate   bool
		withError      error
	}{
		{
			name:           "update consumer",
			consumerConfig: jetstream.ConsumerConfig{Name: "testcons", Description: "updated consumer"},
			shouldUpdate:   true,
		},
		{
			name:           "update consumer, with metadata",
			consumerConfig: jetstream.ConsumerConfig{Name: "testcons", Description: "updated consumer", Metadata: map[string]string{"foo": "bar", "baz": "quux"}},
			shouldUpdate:   true,
		},
		{
			name:           "illegal update",
			consumerConfig: jetstream.ConsumerConfig{Name: "testcons", AckPolicy: jetstream.AckNonePolicy},
			withError:      jetstream.ErrConsumerCreate,
		},
		{
			name:           "consumer does not exist",
			consumerConfig: jetstream.ConsumerConfig{Name: "abc"},
			withError:      jetstream.ErrConsumerDoesNotExist,
		},
	}

	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.CreateConsumer(ctx, jetstream.ConsumerConfig{Name: "testcons"})
		} else {
			_, err = s.CreatePushConsumer(ctx, jetstream.ConsumerConfig{Name: "testcons", DeliverSubject: "inbox"})
		}
		if err != nil {
			t.Fatalf("Unexpected error: %v", err)
		}