RetryConfig.RequestTimeout to configure the timeout of a single reque…

…st (#5503)

* Implement Delivery.Timeout in HTTPMessageSender

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

* Test

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

* Test is passing

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

pkg/kncloudevents/message_sender.go

@@ -57,8 +57,18 @@ func (s *HTTPMessageSender) SendWithRetries(req *nethttp.Request, config *RetryC
 		return s.Send(req)
 	}
 
+	client := s.Client
+	if config.RequestTimeout != 0 {
+		client = &nethttp.Client{
+			Transport:     client.Transport,
+			CheckRedirect: client.CheckRedirect,
+			Jar:           client.Jar,
+			Timeout:       config.RequestTimeout,
+		}
+	}
+
 	retryableClient := retryablehttp.Client{
-		HTTPClient:   s.Client,
+		HTTPClient:   client,
 		RetryWaitMin: defaultRetryWaitMin,
 		RetryWaitMax: defaultRetryWaitMax,
 		RetryMax:     config.RetryMax,

pkg/kncloudevents/message_sender_test.go

@@ -154,6 +154,46 @@ func TestHTTPMessageSenderSendWithRetriesWithBufferedMessage(t *testing.T) {
 	}
 }
 
+func TestHTTPMessageSenderSendWithRetriesWithSingleRequestTimeout(t *testing.T) {
+	t.Parallel()
+
+	timeout := time.Second * 3
+
+	var n int32
+	server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+		newVal := atomic.AddInt32(&n, 1)
+		if newVal >= 5 {
+			writer.WriteHeader(http.StatusOK)
+		} else {
+			// Let's add one more second
+			time.Sleep(timeout)
+			writer.WriteHeader(http.StatusAccepted)
+		}
+	}))
+	defer server.Close()
+
+	sender := &HTTPMessageSender{
+		Client: getClient(),
+	}
+	config := &RetryConfig{
+		RetryMax:   5,
+		CheckRetry: RetryIfGreaterThan300,
+		Backoff: func(attemptNum int, resp *http.Response) time.Duration {
+			return time.Millisecond
+		},
+		RequestTimeout: timeout,
+	}
+
+	request, err := http.NewRequest("POST", server.URL, nil)
+	require.NoError(t, err)
+
+	got, err := sender.SendWithRetries(request, config)
+
+	require.Equal(t, 5, int(atomic.LoadInt32(&n)))
+	require.NoError(t, err)
+	require.Equal(t, http.StatusOK, got.StatusCode)
+}
+
 func TestRetriesOnNetworkErrors(t *testing.T) {
 
 	n := int32(10)

pkg/kncloudevents/retries.go

@@ -63,6 +63,9 @@ type RetryConfig struct {
 
 	CheckRetry CheckRetry
 	Backoff    Backoff
+
+	// RequestTimeout represents the timeout of the single request
+	RequestTimeout time.Duration
 }
 
 func NoRetries() RetryConfig {