Function TestConsumerPinned

jetstream/test/consumer_test.go:552–995

(t *testing.T)

}

func TestConsumerPinned(t *testing.T) {
	t.Run("messages", 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.PriorityPolicyPinned,
			PinnedTTL:      time.Second,
			PriorityGroups: []string{"A"},
		})
		if err != nil {
			t.Fatalf("Unexpected error: %v", err)
		}

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

		gcount := make(chan struct{}, 1000)
		count := atomic.Uint32{}

		// Initially pinned consumer instance
		initiallyPinned, err := s.Consumer(ctx, "cons")
		if err != nil {
			t.Fatalf("Unexpected error: %v", err)
		}

		handler := func(it jetstream.MessagesContext, counter *atomic.Uint32, doneCh chan struct{}) {
			for {
				msg, err := it.Next()
				if err != nil {
					break
				}
				if err := msg.Ack(); err != nil {