HTTP/2: add per-stream idle and lifetime timeouts to core multiplexer

Expose configuration via H2Config, throw H2StreamTimeoutException on expiry and keep the connection alive
Extend test coverage and add an example client demonstrating timed-out and successful streams

Author: arturobernalg

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

@@ -0,0 +1,87 @@
+/*
+ * ====================================================================
+ * 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 java.net.SocketTimeoutException} raised by the HTTP/2 stream
+ * multiplexer when a per-stream timeout elapses.
+ * <p>
+ * This exception is used for timeouts that are scoped to a single HTTP/2
+ * stream rather than the underlying TCP connection, for example:
+ * </p>
+ * <ul>
+ *     <li>an idle timeout where no activity has been observed on the stream, or</li>
+ *     <li>a lifetime timeout where the total age of the stream exceeds
+ *     the configured limit.</li>
+ * </ul>
+ * <p>
+ * The {@link #isIdleTimeout()} flag can be used to distinguish whether
+ * the timeout was triggered by idleness or by the overall stream lifetime.
+ * The affected stream id and the timeout value are exposed via
+ * {@link #getStreamId()} and {@link #getTimeout()} respectively.
+ * </p>
+ *
+ * @since 5.4
+ */
+public class H2StreamTimeoutException extends SocketTimeoutException {
+
+    private static final long serialVersionUID = 1L;
+
+    private final int streamId;
+    private final Timeout timeout;
+    private final boolean idleTimeout;
+
+    public H2StreamTimeoutException(final String message, final int streamId, final Timeout timeout, final boolean idleTimeout) {
+        super(message);
+        this.streamId = streamId;
+        this.timeout = timeout;
+        this.idleTimeout = idleTimeout;
+    }
+
+    public int getStreamId() {
+        return streamId;
+    }
+
+    public Timeout getTimeout() {
+        return timeout;
+    }
+
+    /**
+     * Indicates whether this timeout was triggered by idle time (no activity)
+     * rather than by stream lifetime.
+     *
+     * @return {@code true} if this is an idle timeout, {@code false} if it is a lifetime timeout.
+     */
+    public boolean isIdleTimeout() {
+        return idleTimeout;
+    }
+
+}

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

@@ -75,6 +75,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;
@@ -439,6 +440,10 @@ 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);
                     }
@@ -455,6 +460,9 @@ public final void onInput(final ByteBuffer src) throws HttpException, IOExceptio
                     break;
                 }
             }
+            if (connState.compareTo(ConnectionHandshake.SHUTDOWN) < 0) {
+                checkStreamTimeouts(System.nanoTime());
+            }
         }
     }
 
@@ -532,6 +540,11 @@ public final void onOutput() throws HttpException, IOException {
                 }
             }
         }
+
+        if (connState.compareTo(ConnectionHandshake.SHUTDOWN) < 0) {
+            checkStreamTimeouts(System.nanoTime());
+        }
+
         if (connState.compareTo(ConnectionHandshake.GRACEFUL_SHUTDOWN) == 0) {
             int liveStreams = 0;
             for (final Iterator<H2Stream> it = streams.iterator(); it.hasNext(); ) {
@@ -643,6 +656,7 @@ private void executeRequest(final RequestExecutionCommand requestExecutionComman
                 requestExecutionCommand.getExchangeHandler(),
                 requestExecutionCommand.getPushHandlerFactory(),
                 requestExecutionCommand.getContext()));
+        initializeStreamTimeouts(stream);
 
         if (streamListener != null) {
             final int initInputWindow = stream.getInputWindow().get();
@@ -765,10 +779,12 @@ private void consumeFrame(final RawFrame frame) throws HttpException, IOExceptio
                     final H2StreamChannel channel = createChannel(streamId);
                     if (connState.compareTo(ConnectionHandshake.ACTIVE) <= 0) {
                         stream = streams.createActive(channel, incomingRequest(channel));
+                        initializeStreamTimeouts(stream);
                         streams.resetIfExceedsMaxConcurrentLimit(stream, localConfig.getMaxConcurrentStreams());
                     } else {
                         channel.localReset(H2Error.REFUSED_STREAM);
                         stream = streams.createActive(channel, NoopH2StreamHandler.INSTANCE);
+                        initializeStreamTimeouts(stream);
                     }
                 } else if (stream.isLocalClosed() && stream.isRemoteClosed()) {
                     throw new H2ConnectionException(H2Error.STREAM_CLOSED, "Stream closed");
@@ -959,6 +975,7 @@ private void consumeFrame(final RawFrame frame) throws HttpException, IOExceptio
                     channel.localReset(H2Error.REFUSED_STREAM);
                     promisedStream = streams.createActive(channel, NoopH2StreamHandler.INSTANCE);
                 }
+                initializeStreamTimeouts(promisedStream);
                 try {
                     consumePushPromiseFrame(frame, payload, promisedStream);
                 } catch (final H2StreamResetException ex) {
@@ -1364,8 +1381,18 @@ 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);
+    private void initializeStreamTimeouts(final H2Stream stream) {
+        final Timeout socketTimeout = ioSession.getSocketTimeout();
+        if (socketTimeout != null && socketTimeout.isEnabled()) {
+            stream.setIdleTimeout(socketTimeout);
+            stream.setLifetimeTimeout(socketTimeout);
+        }
+    }
+
+    H2Stream createStream(final H2StreamChannel channel, final H2StreamHandler streamHandler) {
+        final H2Stream stream = streams.createActive(channel, streamHandler);
+        initializeStreamTimeouts(stream);
+        return stream;
     }
 
     private void recordPriorityFromHeaders(final int streamId, final List<? extends Header> headers) {
@@ -1468,6 +1495,7 @@ public void push(final List<Header> headers, final AsyncPushProducer pushProduce
                 final int promisedStreamId = streams.generateStreamId();
                 final H2StreamChannel channel = createChannel(promisedStreamId);
                 final H2Stream stream = streams.createReserved(channel, outgoingPushPromise(channel, pushProducer));
+                initializeStreamTimeouts(stream);
 
                 commitPushPromise(id, promisedStreamId, headers);
                 stream.markRemoteClosed();
@@ -1584,4 +1612,51 @@ 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();
+            final Timeout lifetimeTimeout = stream.getLifetimeTimeout();
+            if ((idleTimeout == null || !idleTimeout.isEnabled())
+                    && (lifetimeTimeout == null || !lifetimeTimeout.isEnabled())) {
+                continue;
+            }
+
+            final long created = stream.getCreatedNanos();
+            final long last = stream.getLastActivityNanos();
+
+            if (idleTimeout != null && idleTimeout.isEnabled()) {
+                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,
+                            true);
+                    stream.localReset(ex, H2Error.CANCEL);
+                    // Once reset due to idle timeout, we do not care about lifetime anymore
+                    continue;
+                }
+            }
+
+            if (lifetimeTimeout != null && lifetimeTimeout.isEnabled()) {
+                final long lifeNanos = lifetimeTimeout.toNanoseconds();
+                if (lifeNanos > 0 && nowNanos - created > lifeNanos) {
+                    final int streamId = stream.getId();
+                    final H2StreamTimeoutException ex = new H2StreamTimeoutException(
+                            "HTTP/2 stream lifetime timeout (" + lifetimeTimeout + ")",
+                            streamId,
+                            lifetimeTimeout,
+                            false);
+                    stream.localReset(ex, H2Error.CANCEL);
+                }
+            }
+        }
+    }
+
 }

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

@@ -43,6 +43,7 @@
 import org.apache.hc.core5.http.nio.HandlerFactory;
 import org.apache.hc.core5.http2.H2Error;
 import org.apache.hc.core5.http2.H2StreamResetException;
+import org.apache.hc.core5.util.Timeout;
 
 class H2Stream {
 
@@ -60,6 +61,12 @@ enum State { RESERVED, OPEN, CLOSED }
     private volatile boolean reserved;
     private volatile boolean remoteClosed;
 
+    private volatile long createdNanos;
+    private volatile long lastActivityNanos;
+
+    private volatile Timeout idleTimeout;
+    private volatile Timeout lifetimeTimeout;
+
     H2Stream(final H2StreamChannel channel, final H2StreamHandler handler, final Consumer<State> stateChangeCallback) {
         this.channel = channel;
         this.handler = handler;
@@ -68,6 +75,8 @@ enum State { RESERVED, OPEN, CLOSED }
         this.transitionRef = new AtomicReference<>(State.RESERVED);
         this.released = new AtomicBoolean();
         this.cancelled = new AtomicBoolean();
+        this.createdNanos = 0L;
+        this.lastActivityNanos = 0L;
     }
 
     int getId() {
@@ -92,6 +101,7 @@ private void triggerClosed() {
 
     void activate() {
         reserved = false;
+        markCreatedAndActive();
         triggerOpen();
     }
 
@@ -134,6 +144,8 @@ boolean isLocalClosed() {
 
     void consumePromise(final List<Header> headers) throws HttpException, IOException {
         try {
+            touch();
+
             if (channel.isLocalReset()) {
                 return;
             }
@@ -153,6 +165,8 @@ void consumeHeader(final List<Header> headers, final boolean endOfStream) throws
             if (endOfStream) {
                 remoteClosed = true;
             }
+            touch();
+
             if (channel.isLocalReset()) {
                 return;
             }
@@ -171,6 +185,8 @@ void consumeData(final ByteBuffer src, final boolean endOfStream) throws HttpExc
             if (endOfStream) {
                 remoteClosed = true;
             }
+            touch();
+
             if (channel.isLocalReset()) {
                 return;
             }
@@ -192,13 +208,16 @@ boolean isOutputReady() {
 
     void produceOutput() throws HttpException, IOException {
         try {
+            touch();
+
             handler.produceOutput();
         } catch (final ProtocolException ex) {
             localReset(ex, H2Error.PROTOCOL_ERROR);
         }
     }
 
     void produceInputCapacityUpdate() throws IOException {
+        touch();
         handler.updateInputCapacity();
     }
 
@@ -289,4 +308,37 @@ public String toString() {
         return buf.toString();
     }
 
+    private void markCreatedAndActive() {
+        final long now = System.nanoTime();
+        this.createdNanos = now;
+        this.lastActivityNanos = now;
+    }
+
+    private void touch() {
+        this.lastActivityNanos = System.nanoTime();
+    }
+
+    long getCreatedNanos() {
+        return createdNanos;
+    }
+
+    long getLastActivityNanos() {
+        return lastActivityNanos;
+    }
+
+    Timeout getIdleTimeout() {
+        return idleTimeout;
+    }
+
+    void setIdleTimeout(final Timeout idleTimeout) {
+        this.idleTimeout = idleTimeout;
+    }
+
+    Timeout getLifetimeTimeout() {
+        return lifetimeTimeout;
+    }
+
+    void setLifetimeTimeout(final Timeout lifetimeTimeout) {
+        this.lifetimeTimeout = lifetimeTimeout;
+    }
 }