Function testJetStreamFetchContext

test/js_test.go:9314–9583

(t *testing.T, srvs ...*jsServer)

}

func testJetStreamFetchContext(t *testing.T, srvs ...*jsServer) {
	srv := srvs[0]
	nc, js := jsClient(t, srv.Server)
	defer nc.Close()

	var err error

	subject := "WQ"
	_, err = js.AddStream(&nats.StreamConfig{
		Name:     subject,
		Replicas: 3,
	})
	if err != nil {
		t.Fatal(err)
	}

	sendMsgs := func(t *testing.T, totalMsgs int) {
		t.Helper()
		for i := 0; i < totalMsgs; i++ {
			payload := fmt.Sprintf("i:%d", i)
			_, err := js.Publish(subject, []byte(payload))
			if err != nil {
				t.Errorf("Unexpected error: %v", err)
			}
		}
	}
	expected := 10
	sendMsgs(t, expected)

	sub, err := js.PullSubscribe(subject, "batch-ctx")
	if err != nil {
		t.Fatal(err)
	}
	defer sub.Unsubscribe()

	t.Run("ctx background", func(t *testing.T) {
		_, err = sub.Fetch(expected, nats.Context(context.Background()))
		if err == nil {
			t.Fatal("Unexpected success")
		}
		if err != nats.ErrNoDeadlineContext {
			t.Errorf("Expected context deadline error, got: %v", err)
		}
	})

	t.Run("ctx canceled", func(t *testing.T) {
		ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
		cancel()

		_, err = sub.Fetch(expected, nats.Context(ctx))
		if err == nil {
			t.Fatal("Unexpected success")
		}
		if err != context.Canceled {
			t.Errorf("Expected context deadline error, got: %v", err)
		}

		ctx, cancel = context.WithCancel(context.Background())