Function TestConsumerOverflow

jetstream/test/consumer_test.go:154–550

(t *testing.T)

}

func TestConsumerOverflow(t *testing.T) {
	t.Run("fetch", func(t *testing.T) {
		srv := RunBasicJetStreamServer()
		defer shutdownJSServerAndRemoveStorage(t, srv)

		nc, err := nats.Connect(srv.ClientURL())
		if err != nil {
			t.Fatalf("Unexpected error: %v", err)
		}
		defer nc.Close()

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

		s, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "foo", Subjects: []string{"FOO.*"}})
		if err != nil {
			t.Fatalf("Unexpected error: %v", err)
		}

		c, err := s.CreateOrUpdateConsumer(ctx, jetstream.ConsumerConfig{
			PriorityPolicy: jetstream.PriorityPolicyOverflow,
			PriorityGroups: []string{"A"},
		})
		if err != nil {
			t.Fatalf("Unexpected error: %v", err)
		}

		// Check that consumer got proper priority policy and TTL
		info := c.CachedInfo()
		if info.Config.PriorityPolicy != jetstream.PriorityPolicyOverflow {
			t.Fatalf("Invalid priority policy; expected: %v; got: %v", jetstream.PriorityPolicyOverflow, info.Config.PriorityPolicy)
		}

		for range 100 {
			_, err = js.Publish(ctx, "FOO.bar", []byte("hello"))
			if err != nil {
				t.Fatalf("Unexpected error: %v", err)
			}
		}

		// We are below overflow, so we should not get any messages.
		msgs, err := c.Fetch(10, jetstream.FetchMinPending(110), jetstream.FetchMaxWait(500*time.Millisecond), jetstream.FetchPriorityGroup("A"))
		if err != nil {
			t.Fatalf("Unexpected error: %v", err)
		}
		count := 0
		for msg := range msgs.Messages() {
			msg.Ack()
			count++
		}
		if count != 0 {
			t.Fatalf("Expected 0 messages, got %d", count)
		}