HTTP/2: per-stream idle timeout

httpcore5-h2/src/main/java/org/apache/hc/core5/http2/H2StreamTimeoutException.java

@@ -0,0 +1,68 @@
+/*
+ * ====================================================================
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ * ====================================================================
+ *
+ * This software consists of voluntary contributions made by many
+ * individuals on behalf of the Apache Software Foundation.  For more
+ * information on the Apache Software Foundation, please see
+ * <http://www.apache.org/>.
+ *
+ */
+package org.apache.hc.core5.http2;
+
+import java.net.SocketTimeoutException;
+
+import org.apache.hc.core5.util.Timeout;
+
+/**
+ * {@link SocketTimeoutException} raised when an HTTP/2 stream exceeds its configured timeout.
+ * <p>
+ * This exception is scoped to a single HTTP/2 stream and is independent of the underlying
+ * connection socket timeout.
+ * </p>
+ * <p>
+ * The stream id and configured timeout value are available via {@link #getStreamId()} and
+ * {@link #getTimeout()}.
+ * </p>
+ *
+ * @since 5.5
+ */
+
+public class H2StreamTimeoutException extends SocketTimeoutException {
+
+    private static final long serialVersionUID = 1L;
+
+    private final int streamId;
+    private final Timeout timeout;
+
+    public H2StreamTimeoutException(final String message, final int streamId, final Timeout timeout) {
+        super(message);
+        this.streamId = streamId;
+        this.timeout = timeout;
+    }
+
+    public int getStreamId() {
+        return streamId;
+    }
+
+    public Timeout getTimeout() {
+        return timeout;
+    }
+
+}

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

@@ -74,6 +74,7 @@
 import org.apache.hc.core5.http2.H2ConnectionException;
 import org.apache.hc.core5.http2.H2Error;
 import org.apache.hc.core5.http2.H2StreamResetException;
+import org.apache.hc.core5.http2.H2StreamTimeoutException;
 import org.apache.hc.core5.http2.config.H2Config;
 import org.apache.hc.core5.http2.config.H2Param;
 import org.apache.hc.core5.http2.config.H2Setting;
@@ -143,6 +144,11 @@ enum SettingsHandshake { READY, TRANSMITTED, ACKED }
     private final Map<Integer, PriorityValue> priorities = new ConcurrentHashMap<>();
     private volatile boolean peerNoRfc7540Priorities;
 
+
+    private static final long STREAM_TIMEOUT_GRANULARITY_MILLIS = 1000;
+    private long lastStreamTimeoutCheckMillis;
+
+
     AbstractH2StreamMultiplexer(
             final ProtocolIOSession ioSession,
             final FrameFactory frameFactory,
@@ -454,6 +460,9 @@ public final void onInput(final ByteBuffer src) throws HttpException, IOExceptio
                     break;
                 }
             }
+            if (connState.compareTo(ConnectionHandshake.SHUTDOWN) < 0) {
+                validateStreamTimeouts();
+            }
         }
     }
 
@@ -531,6 +540,11 @@ public final void onOutput() throws HttpException, IOException {
                 }
             }
         }
+
+        if (connState.compareTo(ConnectionHandshake.SHUTDOWN) < 0) {
+            validateStreamTimeouts();
+        }
+
         if (connState.compareTo(ConnectionHandshake.GRACEFUL_SHUTDOWN) == 0) {
             int liveStreams = 0;
             for (final Iterator<H2Stream> it = streams.iterator(); it.hasNext(); ) {
@@ -1359,8 +1373,9 @@ H2StreamChannel createChannel(final int streamId) {
         return new H2StreamChannelImpl(streamId, initInputWinSize, initOutputWinSize);
     }
 
-    H2Stream createStream(final H2StreamChannel channel, final H2StreamHandler streamHandler) throws H2ConnectionException {
-        return streams.createActive(channel, streamHandler);
+    H2Stream createStream(final H2StreamChannel channel, final H2StreamHandler streamHandler) {
+        final H2Stream stream = streams.createActive(channel, streamHandler);
+        return stream;
     }
 
     private void recordPriorityFromHeaders(final int streamId, final List<? extends Header> headers) {
@@ -1579,4 +1594,37 @@ public String toString() {
 
     }
 
+    private void checkStreamTimeouts(final long nowNanos) throws IOException {
+        for (final Iterator<H2Stream> it = streams.iterator(); it.hasNext(); ) {
+            final H2Stream stream = it.next();
+            if (!stream.isActive()) {
+                continue;
+            }
+
+            final Timeout idleTimeout = stream.getIdleTimeout();
+            if (idleTimeout == null || !idleTimeout.isEnabled()) {
+                continue;
+            }
+
+            final long last = stream.getLastActivityNanos();
+            final long idleNanos = idleTimeout.toNanoseconds();
+            if (idleNanos > 0 && nowNanos - last > idleNanos) {
+                final int streamId = stream.getId();
+                final H2StreamTimeoutException ex = new H2StreamTimeoutException(
+                        "HTTP/2 stream idle timeout (" + idleTimeout + ")",
+                        streamId,
+                        idleTimeout);
+                stream.localReset(ex, H2Error.CANCEL);
+            }
+        }
+    }
+
+    private void validateStreamTimeouts() throws IOException {
+        final long nowMillis = System.currentTimeMillis();
+        if ((nowMillis - lastStreamTimeoutCheckMillis) >= STREAM_TIMEOUT_GRANULARITY_MILLIS) {
+            lastStreamTimeoutCheckMillis = nowMillis;
+            checkStreamTimeouts(System.nanoTime());
+        }
+    }
+
 }

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

@@ -61,6 +61,10 @@ class H2Stream implements StreamControl {
     private volatile boolean reserved;
     private volatile boolean remoteClosed;
 
+    private volatile long lastActivityNanos;
+
+    private volatile Timeout idleTimeout;
+
     H2Stream(final H2StreamChannel channel, final H2StreamHandler handler, final Consumer<State> stateChangeCallback) {
         this.channel = channel;
         this.handler = handler;
@@ -69,6 +73,7 @@ class H2Stream implements StreamControl {
         this.transitionRef = new AtomicReference<>(State.RESERVED);
         this.released = new AtomicBoolean();
         this.cancelled = new AtomicBoolean();
+        this.lastActivityNanos = 0L;
     }
 
     @Override
@@ -83,7 +88,7 @@ public State getState() {
 
     @Override
     public void setTimeout(final Timeout timeout) {
-        // not supported
+        this.idleTimeout = timeout;
     }
 
     boolean isReserved() {
@@ -104,6 +109,7 @@ private void triggerClosed() {
 
     void activate() {
         reserved = false;
+        touch();
         triggerOpen();
     }
 
@@ -146,6 +152,8 @@ boolean isLocalClosed() {
 
     void consumePromise(final List<Header> headers) throws HttpException, IOException {
         try {
+            touch();
+
             if (channel.isLocalReset()) {
                 return;
             }
@@ -165,6 +173,8 @@ void consumeHeader(final List<Header> headers, final boolean endOfStream) throws
             if (endOfStream) {
                 remoteClosed = true;
             }
+            touch();
+
             if (channel.isLocalReset()) {
                 return;
             }
@@ -183,6 +193,8 @@ void consumeData(final ByteBuffer src, final boolean endOfStream) throws HttpExc
             if (endOfStream) {
                 remoteClosed = true;
             }
+            touch();
+
             if (channel.isLocalReset()) {
                 return;
             }
@@ -308,4 +320,16 @@ public String toString() {
         return buf.toString();
     }
 
+    private void touch() {
+        this.lastActivityNanos = System.nanoTime();
+    }
+
+    long getLastActivityNanos() {
+        return lastActivityNanos;
+    }
+
+    Timeout getIdleTimeout() {
+        return idleTimeout;
+    }
+
 }

httpcore5-h2/src/test/java/org/apache/hc/core5/http2/impl/nio/TestAbstractH2StreamMultiplexer.java

@@ -33,6 +33,7 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.Lock;
 import java.util.stream.IntStream;
 
@@ -52,6 +53,7 @@
 import org.apache.hc.core5.http2.H2ConnectionException;
 import org.apache.hc.core5.http2.H2Error;
 import org.apache.hc.core5.http2.H2StreamResetException;
+import org.apache.hc.core5.http2.H2StreamTimeoutException;
 import org.apache.hc.core5.http2.WritableByteChannelMock;
 import org.apache.hc.core5.http2.config.H2Config;
 import org.apache.hc.core5.http2.config.H2Param;
@@ -65,6 +67,7 @@
 import org.apache.hc.core5.http2.hpack.HPackEncoder;
 import org.apache.hc.core5.reactor.ProtocolIOSession;
 import org.apache.hc.core5.util.ByteArrayBuffer;
+import org.apache.hc.core5.util.Timeout;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -1034,5 +1037,51 @@ void testPriorityUpdateContinuesAfterSettingsWithNoH2Equals1() throws Exception
         Assertions.assertTrue(idxPriUpd >= 0, "PRIORITY_UPDATE should be emitted when NO_RFC7540=1");
     }
 
+    @Test
+    void testStreamIdleTimeoutTriggersH2StreamTimeoutException() throws Exception {
+        Mockito.when(protocolIOSession.write(ArgumentMatchers.any(ByteBuffer.class)))
+                .thenAnswer(invocation -> {
+                    final ByteBuffer buffer = invocation.getArgument(0, ByteBuffer.class);
+                    final int remaining = buffer.remaining();
+                    buffer.position(buffer.limit());
+                    return remaining;
+                });
+        Mockito.doNothing().when(protocolIOSession).setEvent(ArgumentMatchers.anyInt());
+        Mockito.doNothing().when(protocolIOSession).clearEvent(ArgumentMatchers.anyInt());
+
+        final H2Config h2Config = H2Config.custom().build();
+        final AbstractH2StreamMultiplexer streamMultiplexer = new H2StreamMultiplexerImpl(
+                protocolIOSession,
+                FRAME_FACTORY,
+                StreamIdGenerator.ODD,
+                httpProcessor,
+                CharCodingConfig.DEFAULT,
+                h2Config,
+                h2StreamListener,
+                () -> streamHandler);
+
+        final H2StreamChannel channel = streamMultiplexer.createChannel(1);
+        final H2Stream stream = streamMultiplexer.createStream(channel, streamHandler);
+
+        stream.setTimeout(Timeout.of(1, TimeUnit.NANOSECONDS));
+        stream.activate();
+
+        streamMultiplexer.onOutput();
+
+        Mockito.verify(streamHandler).failed(exceptionCaptor.capture());
+        final Exception cause = exceptionCaptor.getValue();
+        Assertions.assertInstanceOf(H2StreamTimeoutException.class, cause);
+
+        final H2StreamTimeoutException timeoutEx = (H2StreamTimeoutException) cause;
+        Assertions.assertEquals(1, timeoutEx.getStreamId());
+
+        Assertions.assertTrue(stream.isLocalClosed());
+        Assertions.assertTrue(stream.isClosed());
+
+        Assertions.assertTrue(timeoutEx.getMessage().contains("idle timeout"));
+        Assertions.assertEquals(1L, timeoutEx.getTimeout().toNanoseconds());
+        Assertions.assertEquals(1, timeoutEx.getStreamId());
+
+    }
 }