Function testContextRequestWithCancel

test/context_test.go:211–288

(t *testing.T, nc *nats.Conn)

}

func testContextRequestWithCancel(t *testing.T, nc *nats.Conn) {
	ctx, cancelCB := context.WithCancel(context.Background())
	defer cancelCB() // should always be called, not discarded, to prevent context leak

	// timer which cancels the context though can also be arbitrarily extended
	expirationTimer := time.AfterFunc(100*time.Millisecond, func() {
		cancelCB()
	})

	sub1, err := nc.Subscribe("slow", func(m *nats.Msg) {
		// simulates latency into the client so that timeout is hit.
		time.Sleep(40 * time.Millisecond)
		nc.Publish(m.Reply, []byte("OK"))
	})
	if err != nil {
		t.Fatalf("Expected to be able to subscribe: %s", err)
	}
	defer sub1.Unsubscribe()
	sub2, err := nc.Subscribe("slower", func(m *nats.Msg) {
		// we know this request will take longer so extend the timeout
		expirationTimer.Reset(100 * time.Millisecond)

		// slower reply which would have hit original timeout
		time.Sleep(70 * time.Millisecond)

		nc.Publish(m.Reply, []byte("Also OK"))
	})
	if err != nil {
		t.Fatalf("Expected to be able to subscribe: %s", err)
	}
	defer sub2.Unsubscribe()

	for i := 0; i < 2; i++ {
		resp, err := nc.RequestWithContext(ctx, "slow", []byte(""))
		if err != nil {
			t.Fatalf("Expected request with context to not fail: %s", err)
		}
		got := string(resp.Data)
		expected := "OK"
		if got != expected {
			t.Errorf("Expected to receive %s, got: %s", expected, got)
		}
	}

	// A third request with latency would make the context
	// get canceled, but these reset the timer so deadline
	// gets extended:
	for i := 0; i < 10; i++ {
		resp, err := nc.RequestWithContext(ctx, "slower", []byte(""))
		if err != nil {
			t.Fatalf("Expected request with context to not fail: %s", err)
		}
		got := string(resp.Data)
		expected := "Also OK"
		if got != expected {
			t.Errorf("Expected to receive %s, got: %s", expected, got)
		}
	}