Function TestJetStreamSubscribeContextCancel

test/js_test.go:9585–9670

(t *testing.T)

}

func TestJetStreamSubscribeContextCancel(t *testing.T) {
	s := RunBasicJetStreamServer()
	defer shutdownJSServerAndRemoveStorage(t, s)

	nc, js := jsClient(t, s)
	defer nc.Close()

	var err error

	// Create the stream using our client API.
	_, err = js.AddStream(&nats.StreamConfig{
		Name:     "TEST",
		Subjects: []string{"foo", "bar", "baz", "foo.*"},
	})
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}

	toSend := 100
	for i := 0; i < toSend; i++ {
		js.Publish("bar", []byte("foo"))
	}

	t.Run("cancel unsubscribes and deletes ephemeral", func(t *testing.T) {
		ctx, cancel := context.WithCancel(context.Background())
		defer cancel()

		ch := make(chan *nats.Msg, 100)
		sub, err := js.Subscribe("bar", func(msg *nats.Msg) {
			ch <- msg

			// Cancel will unsubscribe and remove the subscription
			// of the consumer.
			if len(ch) >= 50 {
				cancel()
			}
		}, nats.Context(ctx))
		if err != nil {
			t.Fatal(err)
		}

		select {
		case <-ctx.Done():
		case <-time.After(3 * time.Second):
			t.Fatal("Timed out waiting for context to be canceled")
		}

		// Consumer should not be present since unsubscribe already called.
		checkFor(t, 2*time.Second, 15*time.Millisecond, func() error {
			info, err := sub.ConsumerInfo()
			if err != nil && err == nats.ErrConsumerNotFound {
				return nil
			}
			return fmt.Errorf("Consumer still active, got: %v (info=%+v)", err, info)
		})

		got := len(ch)
		expected := 50