Add core async HTTP/2 request-timeout test asserting HttpStreamResetException

Author: arturobernalg

httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/AbstractH2StreamMultiplexer.java

@@ -30,6 +30,7 @@
 import java.net.SocketAddress;
 import java.nio.BufferOverflowException;
 import java.nio.ByteBuffer;
+import java.nio.channels.CancelledKeyException;
 import java.nio.channels.SelectionKey;
 import java.nio.charset.StandardCharsets;
 import java.util.Deque;
@@ -271,7 +272,12 @@ private void commitFrameInternal(final RawFrame frame) throws IOException {
         } else {
             outputQueue.addLast(frame);
         }
-        ioSession.setEvent(SelectionKey.OP_WRITE);
+        try {
+            ioSession.setEvent(SelectionKey.OP_WRITE);
+        } catch (final CancelledKeyException ex) {
+            connState = ConnectionHandshake.SHUTDOWN;
+            ioSession.close(CloseMode.IMMEDIATE);
+        }
     }
 
     private void commitFrame(final RawFrame frame) throws IOException {
@@ -413,7 +419,12 @@ private void incrementInputCapacity(
 
     void requestSessionOutput() {
         outputRequests.incrementAndGet();
-        ioSession.setEvent(SelectionKey.OP_WRITE);
+        try {
+            ioSession.setEvent(SelectionKey.OP_WRITE);
+        } catch (final CancelledKeyException ex) {
+            connState = ConnectionHandshake.SHUTDOWN;
+            ioSession.close(CloseMode.IMMEDIATE);
+        }
     }
 
     public final void onConnect() throws HttpException, IOException {
@@ -439,10 +450,6 @@ public final void onInput(final ByteBuffer src) throws HttpException, IOExceptio
             for (;;) {
                 final RawFrame frame = inputBuffer.read(src, ioSession);
                 if (frame != null) {
-                    if (connState.compareTo(ConnectionHandshake.SHUTDOWN) < 0) {
-                        checkStreamTimeouts(System.nanoTime());
-                    }
-
                     if (streamListener != null) {
                         streamListener.onFrameInput(this, frame.getStreamId(), frame);
                     }
@@ -1377,9 +1384,12 @@ H2StreamChannel createChannel(final int streamId) {
     }
 
     private void initializeStreamTimeouts(final H2Stream stream) {
-        final Timeout socketTimeout = ioSession.getSocketTimeout();
-        if (socketTimeout != null && socketTimeout.isEnabled()) {
-            stream.setIdleTimeout(socketTimeout);
+        final Timeout streamIdleTimeout = stream.getIdleTimeout();
+        if (streamIdleTimeout == null || !streamIdleTimeout.isEnabled()) {
+            final Timeout socketTimeout = ioSession.getSocketTimeout();
+            if (socketTimeout != null && socketTimeout.isEnabled()) {
+                stream.setIdleTimeout(socketTimeout);
+            }
         }
     }
 
@@ -1633,7 +1643,6 @@ private void checkStreamTimeouts(final long nowNanos) throws IOException {
                             idleTimeout,
                             true);
                     stream.localReset(ex, H2Error.CANCEL);
-                    // Once reset due to idle timeout, we do not care about lifetime anymore
                     continue;
                 }
             }

httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/H2Stream.java

@@ -213,6 +213,13 @@ boolean isOutputReady() {
 
     void produceOutput() throws HttpException, IOException {
         try {
+            if (channel.isLocalReset()) {
+                return;
+            }
+            if (cancelled.get()) {
+                localResetCancelled();
+                return;
+            }
             touch();
 
             handler.produceOutput();
@@ -222,6 +229,13 @@ void produceOutput() throws HttpException, IOException {
     }
 
     void produceInputCapacityUpdate() throws IOException {
+        if (channel.isLocalReset()) {
+            return;
+        }
+        if (cancelled.get()) {
+            localResetCancelled();
+            return;
+        }
         touch();
         handler.updateInputCapacity();
     }

httpcore5-h2/src/test/java/org/apache/hc/core5/http2/examples/H2StreamTimeoutClientExample.java

@@ -32,13 +32,16 @@
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.apache.hc.core5.http.EntityDetails;
 import org.apache.hc.core5.http.Header;
 import org.apache.hc.core5.http.HttpConnection;
 import org.apache.hc.core5.http.HttpException;
+import org.apache.hc.core5.http.HttpHost;
 import org.apache.hc.core5.http.HttpResponse;
 import org.apache.hc.core5.http.impl.bootstrap.HttpAsyncRequester;
+import org.apache.hc.core5.http.nio.AsyncClientEndpoint;
 import org.apache.hc.core5.http.nio.AsyncClientExchangeHandler;
 import org.apache.hc.core5.http.nio.AsyncRequestProducer;
 import org.apache.hc.core5.http.nio.CapacityChannel;
@@ -50,6 +53,7 @@
 import org.apache.hc.core5.http.protocol.HttpContext;
 import org.apache.hc.core5.http.protocol.HttpCoreContext;
 import org.apache.hc.core5.http2.H2StreamResetException;
+import org.apache.hc.core5.http2.H2StreamTimeoutException;
 import org.apache.hc.core5.http2.HttpVersionPolicy;
 import org.apache.hc.core5.http2.config.H2Config;
 import org.apache.hc.core5.http2.frame.RawFrame;
@@ -60,23 +64,13 @@
 import org.apache.hc.core5.util.Timeout;
 
 /**
- * Example of an HTTP/2 client where a "slow" request gets aborted when the
- * underlying HTTP/2 connection times out due to inactivity (socket timeout).
+ * Example of an HTTP/2 client where a "slow" request gets aborted by a
+ * per-stream idle timeout enforced by the HTTP/2 multiplexer.
  * <p>
- * The client opens a single HTTP/2 connection to {@code nghttp2.org} and
- * executes two concurrent requests:
- * <ul>
- *     <li>a "fast" request ({@code /httpbin/ip}), which completes before
- *     the connection idle timeout, and</li>
- *     <li>a "slow" request ({@code /httpbin/delay/5}), which keeps the
- *     connection idle long enough for the I/O reactor to trigger a timeout
- *     and close the HTTP/2 connection.</li>
- * </ul>
- * <p>
- * When the reactor closes the connection due to inactivity, all active
- * streams fail with {@link H2StreamResetException} reporting
- * {@code "Timeout due to inactivity (...)"}. The already completed stream
- * is not affected.
+ * The connection socket timeout is set to 2 seconds and is used as the initial / default
+ * per-stream idle timeout value. The example keeps the connection active by sending
+ * small "keep-alive" requests on separate streams, so the connection itself does not time out.
+ * The "slow" stream remains idle long enough to exceed the per-stream idle timeout and gets reset.
  *
  * @since 5.4
  */
@@ -85,8 +79,6 @@ public class H2StreamTimeoutClientExample {
     public static void main(final String[] args) throws Exception {
 
         final IOReactorConfig ioReactorConfig = IOReactorConfig.custom()
-                // Connection-level inactivity timeout: keep it short so that
-                // /httpbin/delay/5 reliably triggers it.
                 .setSoTimeout(2, TimeUnit.SECONDS)
                 .build();
 
@@ -167,52 +159,150 @@ public void onOutputFlowControl(
 
         requester.start();
 
-        final URI fastUri = new URI("https://nghttp2.org/httpbin/ip");
-        final URI slowUri = new URI("https://nghttp2.org/httpbin/delay/5");
+        final HttpHost target = new HttpHost("https", "nghttp2.org", 443);
+        final AsyncClientEndpoint endpoint = requester.connect(target, Timeout.ofSeconds(10)).get();
+
+        try {
+            final URI keepAliveUri = new URI("https://nghttp2.org/httpbin/ip");
+            final URI slowUri = new URI("https://nghttp2.org/httpbin/delay/5");
+
+            final CountDownLatch latch = new CountDownLatch(2);
+            final AtomicBoolean stop = new AtomicBoolean(false);
+
+            // Keep the connection active with short requests on new streams,
+            // so the connection does NOT hit "Timeout due to inactivity".
+            final Thread keepAliveThread = new Thread(() -> {
+                try {
+                    while (!stop.get()) {
+                        executeKeepAliveOnce(endpoint, keepAliveUri);
+                        Thread.sleep(500);
+                    }
+                } catch (final Exception ignore) {
+                } finally {
+                    latch.countDown();
+                }
+            });
+            keepAliveThread.setDaemon(true);
+            keepAliveThread.start();
+
+            // Slow stream: should be reset by per-stream idle timeout while the connection stays active.
+            executeWithLogging(
+                    endpoint,
+                    slowUri,
+                    "[slow]",
+                    latch,
+                    stop);
+
+            latch.await(30, TimeUnit.SECONDS);
+
+        } finally {
+            endpoint.releaseAndReuse();
+            System.out.println("Shutting down I/O reactor");
+            requester.initiateShutdown();
+        }
+    }
+
+    private static void executeKeepAliveOnce(
+            final AsyncClientEndpoint endpoint,
+            final URI requestUri) throws InterruptedException {
+
+        final AsyncRequestProducer requestProducer = AsyncRequestBuilder.get(requestUri).build();
+        final BasicResponseConsumer<String> responseConsumer = new BasicResponseConsumer<>(
+                new StringAsyncEntityConsumer());
+
+        final CountDownLatch done = new CountDownLatch(1);
+
+        endpoint.execute(new AsyncClientExchangeHandler() {
+
+            @Override
+            public void releaseResources() {
+                requestProducer.releaseResources();
+                responseConsumer.releaseResources();
+                done.countDown();
+            }
+
+            @Override
+            public void cancel() {
+                done.countDown();
+            }
+
+            @Override
+            public void failed(final Exception cause) {
+                done.countDown();
+            }
+
+            @Override
+            public void produceRequest(
+                    final RequestChannel channel,
+                    final HttpContext httpContext) throws HttpException, IOException {
+                requestProducer.sendRequest(channel, httpContext);
+            }
+
+            @Override
+            public int available() {
+                return requestProducer.available();
+            }
+
+            @Override
+            public void produce(final DataStreamChannel channel) throws IOException {
+                requestProducer.produce(channel);
+            }
+
+            @Override
+            public void consumeInformation(
+                    final HttpResponse response,
+                    final HttpContext httpContext) throws HttpException, IOException {
+                // No-op
+            }
+
+            @Override
+            public void consumeResponse(
+                    final HttpResponse response,
+                    final EntityDetails entityDetails,
+                    final HttpContext httpContext) throws HttpException, IOException {
+                responseConsumer.consumeResponse(response, entityDetails, httpContext, null);
+            }
 
-        final CountDownLatch latch = new CountDownLatch(2);
+            @Override
+            public void updateCapacity(final CapacityChannel capacityChannel) throws IOException {
+                responseConsumer.updateCapacity(capacityChannel);
+            }
 
-        // --- Fast stream: expected to succeed
-        executeWithLogging(
-                requester,
-                fastUri,
-                "[fast]",
-                latch,
-                false);
+            @Override
+            public void consume(final ByteBuffer src) throws IOException {
+                responseConsumer.consume(src);
+            }
 
-        // --- Slow stream: /delay/5 sleeps 5 seconds and should exceed
-        // the 2-second connection idle timeout, resulting in a reset.
-        executeWithLogging(
-                requester,
-                slowUri,
-                "[slow]",
-                latch,
-                true);
+            @Override
+            public void streamEnd(final List<? extends Header> trailers)
+                    throws HttpException, IOException {
+                responseConsumer.streamEnd(trailers);
+            }
 
-        latch.await();
+        }, HttpCoreContext.create());
 
-        System.out.println("Shutting down I/O reactor");
-        requester.initiateShutdown();
+        done.await(5, TimeUnit.SECONDS);
     }
 
     private static void executeWithLogging(
-            final HttpAsyncRequester requester,
+            final AsyncClientEndpoint endpoint,
             final URI requestUri,
             final String label,
             final CountDownLatch latch,
-            final boolean expectTimeout) {
+            final AtomicBoolean stop) {
 
         final AsyncRequestProducer requestProducer = AsyncRequestBuilder.get(requestUri)
                 .build();
         final BasicResponseConsumer<String> responseConsumer = new BasicResponseConsumer<>(
                 new StringAsyncEntityConsumer());
 
-        requester.execute(new AsyncClientExchangeHandler() {
+        endpoint.execute(new AsyncClientExchangeHandler() {
 
             @Override
             public void releaseResources() {
                 requestProducer.releaseResources();
                 responseConsumer.releaseResources();
+                stop.set(true);
                 latch.countDown();
             }
 
@@ -223,11 +313,12 @@ public void cancel() {
 
             @Override
             public void failed(final Exception cause) {
-                if (expectTimeout && cause instanceof H2StreamResetException) {
-                    final H2StreamResetException ex = (H2StreamResetException) cause;
-                    System.out.println(label + " expected timeout reset: "
-                            + requestUri
-                            + " -> " + ex);
+                if (cause instanceof H2StreamTimeoutException) {
+                    System.out.println(label + " expected per-stream timeout reset: "
+                            + requestUri + " -> " + cause);
+                } else if (cause instanceof H2StreamResetException) {
+                    System.out.println(label + " stream reset: "
+                            + requestUri + " -> " + cause);
                 } else {
                     System.out.println(label + " failure: "
                             + requestUri + " -> " + cause);
@@ -265,13 +356,8 @@ public void consumeResponse(
                     final HttpResponse response,
                     final EntityDetails entityDetails,
                     final HttpContext httpContext) throws HttpException, IOException {
-                if (expectTimeout) {
-                    System.out.println(label + " UNEXPECTED success: "
-                            + requestUri + " -> " + response.getCode());
-                } else {
-                    System.out.println(label + " response: "
-                            + requestUri + " -> " + response.getCode());
-                }
+                System.out.println(label + " response: "
+                        + requestUri + " -> " + response.getCode());
                 responseConsumer.consumeResponse(response, entityDetails, httpContext, null);
             }
 
@@ -289,12 +375,9 @@ public void consume(final ByteBuffer src) throws IOException {
             public void streamEnd(final List<? extends Header> trailers)
                     throws HttpException, IOException {
                 responseConsumer.streamEnd(trailers);
-                if (!expectTimeout) {
-                    System.out.println(label + " body completed for " + requestUri);
-                }
             }
 
-        }, Timeout.ofSeconds(10), HttpCoreContext.create());
+        }, HttpCoreContext.create());
     }
 
 }