Function TestAsyncResponsesDoesNotLeak

modules/frontend/pipeline/responses_test.go:215–403

(t *testing.T)

}

func TestAsyncResponsesDoesNotLeak(t *testing.T) {
	tcs := []struct {
		name    string
		finalRT func(requestCancel context.CancelFunc) RoundTripperFunc
		cleanup func()
	}{
		{
			name: "happy path",
			finalRT: func(_ context.CancelFunc) RoundTripperFunc {
				return RoundTripperFunc(func(_ Request) (*http.Response, error) {
					return &http.Response{
						Body: io.NopCloser(strings.NewReader("foo")),
					}, nil
				})
			},
		},
		{
			name: "error path",
			finalRT: func(_ context.CancelFunc) RoundTripperFunc {
				return RoundTripperFunc(func(_ Request) (*http.Response, error) {
					return nil, errors.New("foo")
				})
			},
		},
		{
			name: "combiner bails early",
			finalRT: func(_ context.CancelFunc) RoundTripperFunc {
				responseCounter := atomic.Int32{}

				return RoundTripperFunc(func(_ Request) (*http.Response, error) {
					counter := responseCounter.Add(1)
					if counter == 2 {
						return &http.Response{
							StatusCode: http.StatusNotFound, // force the combiner to bail on the second response
							Body:       io.NopCloser(strings.NewReader("foo")),
						}, nil
					}

					return &http.Response{
						Body: io.NopCloser(strings.NewReader("foo")),
					}, nil
				})
			},
		},
		{
			name: "context cancelled before returned responses",
			finalRT: func(cancel context.CancelFunc) RoundTripperFunc {
				go func() {
					time.Sleep(1 * time.Second)
					cancel()
				}()

				return RoundTripperFunc(func(_ Request) (*http.Response, error) {
					time.Sleep(3 * time.Second)

					return &http.Response{
						Body: io.NopCloser(strings.NewReader("foo")),
					}, nil