Function TestConsumerPrioritized

jetstream/test/consumer_test.go:1161–1438

(t *testing.T)

}

func TestConsumerPrioritized(t *testing.T) {
	t.Run("consume", 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{
			Durable:        "cons",
			AckPolicy:      jetstream.AckExplicitPolicy,
			Description:    "test consumer",
			PriorityPolicy: jetstream.PriorityPolicyPrioritized,
			PriorityGroups: []string{"A"},
		})
		if err != nil {
			t.Fatalf("Unexpected error: %v", err)
		}
		// Check that consumer got proper priority policy
		info := c.CachedInfo()
		if info.Config.PriorityPolicy != jetstream.PriorityPolicyPrioritized {
			t.Fatalf("Invalid priority policy; expected: %v; got: %v", jetstream.PriorityPolicyPrioritized, info.Config.PriorityPolicy)
		}
		cons1, err := s.Consumer(ctx, "cons")
		if err != nil {
			t.Fatalf("Unexpected error: %v", err)
		}
		cons2, err := s.Consumer(ctx, "cons")
		if err != nil {
			t.Fatalf("Unexpected error: %v", err)
		}

		wg := sync.WaitGroup{}
		wg.Add(100)
		errs := make(chan error, 100)
		cc1, err := cons1.Consume(func(m jetstream.Msg) {
			if err := m.Ack(); err != nil {
				errs <- fmt.Errorf("Failed to ack message: %v", err)
			}
			wg.Done()
		}, jetstream.PullPriorityGroup("A"), jetstream.PullPrioritized(0), jetstream.PullMaxMessages(10))
		if err != nil {
			t.Fatalf("Unexpected error: %v", err)