Function TestConsumerCachedInfo

jetstream/test/consumer_test.go:1440–1525

(t *testing.T)

}

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

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	js, err := jetstream.New(nc)
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}
	defer nc.Close()

	for _, consType := range []string{"pull", "push"} {
		t.Run(consType, func(t *testing.T) {
			s, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "foo", Subjects: []string{"FOO.*"}})
			if err != nil {
				t.Fatalf("Unexpected error: %v", err)
			}
			var c consInfoConstraint
			if consType == "pull" {
				c, err = s.CreateOrUpdateConsumer(ctx, jetstream.ConsumerConfig{
					Durable:     "cons",
					AckPolicy:   jetstream.AckExplicitPolicy,
					Description: "test consumer",
				})
			} else {
				c, err = s.CreateOrUpdatePushConsumer(ctx, jetstream.ConsumerConfig{
					DeliverSubject: "inbox",
					Durable:        "cons",
					AckPolicy:      jetstream.AckExplicitPolicy,
					Description:    "test consumer",
				})
			}
			if err != nil {
				t.Fatalf("Unexpected error: %v", err)
			}

			info := c.CachedInfo()

			if info.Stream != "foo" {
				t.Fatalf("Invalid stream name; expected: 'foo'; got: %s", info.Stream)
			}
			if info.Config.Description != "test consumer" {
				t.Fatalf("Invalid consumer description; expected: 'test consumer'; got: %s", info.Config.Description)
			}

			// update consumer and see if info is updated
			if consType == "pull" {
				_, err = s.CreateOrUpdateConsumer(ctx, jetstream.ConsumerConfig{
					Durable:     "cons",
					AckPolicy:   jetstream.AckExplicitPolicy,
					Description: "updated consumer",
				})
			} else {