Function TestConsumerInfo

jetstream/test/consumer_test.go:29–152

(t *testing.T)

)

func TestConsumerInfo(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"} {
		s, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "foo", Subjects: []string{"FOO.*"}})
		if err != nil {
			t.Fatalf("Unexpected error: %v", err)
		}

		t.Run(fmt.Sprintf("get %s consumer info, ok", consType), func(t *testing.T) {
			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.CreatePushConsumer(ctx, jetstream.ConsumerConfig{
					DeliverSubject: "inbox",
					Durable:        "cons",
					AckPolicy:      jetstream.AckExplicitPolicy,
					Description:    "test consumer",
				})
			}
			if err != nil {
				t.Fatalf("Unexpected error: %v", err)
			}

			info, err := c.Info(ctx)
			if err != nil {
				t.Fatalf("Unexpected error: %v", err)
			}

			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)
			}
			if info.Config.PauseUntil != nil {
				t.Fatalf("Consumer should not be paused")
			}
			if info.Paused != false {