Function TestConsumerUnpin

jetstream/test/consumer_test.go:997–1159

(t *testing.T)

}

func TestConsumerUnpin(t *testing.T) {
	t.Run("unpin consumer", 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:      50 * 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)
			}
		}

		msgs, err := c.Messages(jetstream.PullPriorityGroup("A"))
		if err != nil {
			t.Fatalf("Unexpected error: %v", err)
		}
		defer msgs.Stop()

		msg, err := msgs.Next()
		if err != nil {
			t.Fatalf("Unexpected error: %v", err)
		}

		firstPinID := msg.Headers().Get("Nats-Pin-Id")
		if firstPinID == "" {
			t.Fatalf("Expected pinned message")
		}