Apply response timeout to H2 streams

Author: ok2c

httpclient5/src/main/java/org/apache/hc/client5/http/impl/async/InternalH2AsyncExecRuntime.java

@@ -297,17 +297,25 @@ public Cancellable execute(
                 : exchangeHandler;
         final ComplexCancellable complexCancellable = new ComplexCancellable();
         final IOSession session = endpoint.session;
+        final RequestConfig requestConfig = context.getRequestConfigOrDefault();
+        final Timeout responseTimeout = requestConfig.getResponseTimeout();
         if (session.isOpen()) {
             if (log.isDebugEnabled()) {
                 log.debug("{} start execution {}", ConnPoolSupport.getId(endpoint), id);
             }
             context.setProtocolVersion(HttpVersion.HTTP_2);
             session.enqueue(
-                    new RequestExecutionCommand(actual, pushHandlerFactory, complexCancellable, context),
+                    new RequestExecutionCommand(
+                            actual,
+                            pushHandlerFactory,
+                            context,
+                            streamControl -> {
+                                streamControl.setTimeout(responseTimeout);
+                                complexCancellable.setDependency(streamControl);
+                            }),
                     Command.Priority.NORMAL);
         } else {
             final HttpRoute route = endpoint.route;
-            final RequestConfig requestConfig = context.getRequestConfigOrDefault();
             @SuppressWarnings("deprecation")
             final Timeout connectTimeout = requestConfig.getConnectTimeout();
             connPool.getSession(route, connectTimeout, new FutureCallback<IOSession>() {
@@ -321,7 +329,14 @@ public void completed(final IOSession ioSession) {
                     }
                     context.setProtocolVersion(HttpVersion.HTTP_2);
                     ioSession.enqueue(
-                            new RequestExecutionCommand(actual, pushHandlerFactory, complexCancellable, context),
+                            new RequestExecutionCommand(
+                                    actual,
+                                    pushHandlerFactory,
+                                    context,
+                                    streamControl -> {
+                                        streamControl.setTimeout(responseTimeout);
+                                        complexCancellable.setDependency(streamControl);
+                                    }),
                             Command.Priority.NORMAL);
                 }
 

httpclient5/src/main/java/org/apache/hc/client5/http/impl/async/MinimalH2AsyncClient.java

@@ -142,6 +142,7 @@ public Cancellable execute(
                 }
                 @SuppressWarnings("deprecation")
                 final Timeout connectTimeout = requestConfig.getConnectTimeout();
+                final Timeout responseTimeout = requestConfig.getResponseTimeout();
                 final HttpHost target = new HttpHost(request.getScheme(), request.getAuthority());
 
                 final Future<IOSession> sessionFuture = connPool.getSession(new HttpRoute(target), connectTimeout,
@@ -224,16 +225,22 @@ public void streamEnd(final List<? extends Header> trailers) throws HttpExceptio
                                     new RequestExecutionCommand(
                                             new LoggingAsyncClientExchangeHandler(LOG, exchangeId, internalExchangeHandler),
                                             pushHandlerFactory,
-                                            cancellable,
-                                            clientContext),
+                                            clientContext,
+                                            streamControl -> {
+                                                streamControl.setTimeout(responseTimeout);
+                                                cancellable.setDependency(streamControl);
+                                            }),
                                     Command.Priority.NORMAL);
                         } else {
                             session.enqueue(
                                     new RequestExecutionCommand(
                                             internalExchangeHandler,
                                             pushHandlerFactory,
-                                            cancellable,
-                                            clientContext),
+                                            clientContext,
+                                            streamControl -> {
+                                                streamControl.setTimeout(responseTimeout);
+                                                cancellable.setDependency(streamControl);
+                                            }),
                                     Command.Priority.NORMAL);
                         }
                     }

httpclient5/src/main/java/org/apache/hc/client5/http/impl/nio/PoolingAsyncClientConnectionManager.java

@@ -50,6 +50,7 @@
 import org.apache.hc.client5.http.nio.AsyncClientConnectionOperator;
 import org.apache.hc.client5.http.nio.AsyncConnectionEndpoint;
 import org.apache.hc.client5.http.nio.ManagedAsyncClientConnection;
+import org.apache.hc.client5.http.protocol.HttpClientContext;
 import org.apache.hc.client5.http.ssl.DefaultClientTlsStrategy;
 import org.apache.hc.core5.annotation.Contract;
 import org.apache.hc.core5.annotation.Internal;
@@ -819,7 +820,15 @@ public void execute(
             }
             context.setProtocolVersion(connection.getProtocolVersion());
             connection.submitCommand(
-                    new RequestExecutionCommand(exchangeHandler, pushHandlerFactory, context),
+                    new RequestExecutionCommand(
+                            exchangeHandler,
+                            pushHandlerFactory,
+                            context,
+                            streamControl -> {
+                                final HttpClientContext clientContext = HttpClientContext.cast(context);
+                                final Timeout responseTimeout = clientContext.getRequestConfigOrDefault().getResponseTimeout();
+                                streamControl.setTimeout(responseTimeout);
+                            }),
                     Command.Priority.NORMAL);
         }