add per-stream idle timeout and suppress CancelledKeyException on cancel

;

Author: arturobernalg

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

@@ -26,7 +26,7 @@
  */
 package org.apache.hc.core5.http2;
 
-import java.net.SocketTimeoutException;
+import java.io.InterruptedIOException;
 
 import org.apache.hc.core5.util.Timeout;
 
@@ -51,7 +51,7 @@
  *
  * @since 5.4
  */
-public class H2StreamTimeoutException extends SocketTimeoutException {
+public class H2StreamTimeoutException extends InterruptedIOException {
 
     private static final long serialVersionUID = 1L;
 

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

@@ -439,10 +439,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);
                     }
@@ -655,7 +651,6 @@ private void executeRequest(final RequestExecutionCommand requestExecutionComman
                 requestExecutionCommand.getExchangeHandler(),
                 requestExecutionCommand.getPushHandlerFactory(),
                 requestExecutionCommand.getContext()));
-        initializeStreamTimeouts(stream);
 
         if (streamListener != null) {
             final int initInputWindow = stream.getInputWindow().get();
@@ -1377,9 +1372,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);
+            }
         }
     }
 
@@ -1614,43 +1612,21 @@ private void checkStreamTimeouts(final long nowNanos) throws IOException {
             }
 
             final Timeout idleTimeout = stream.getIdleTimeout();
-            final Timeout lifetimeTimeout = stream.getLifetimeTimeout();
-            if ((idleTimeout == null || !idleTimeout.isEnabled())
-                    && (lifetimeTimeout == null || !lifetimeTimeout.isEnabled())) {
+            if (idleTimeout == null || !idleTimeout.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);
-                }
+            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);
             }
         }
     }
-
 }

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

@@ -218,6 +218,13 @@ boolean isOutputReady() {
 
     void produceOutput() throws HttpException, IOException {
         try {
+            if (channel.isLocalReset()) {
+                return;
+            }
+            if (cancelled.get()) {
+                localResetCancelled();
+                return;
+            }
             touch();
 
             handler.produceOutput();
@@ -227,6 +234,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/main/java/org/apache/hc/core5/http2/impl/nio/bootstrap/CancellableExecution.java

@@ -26,6 +26,7 @@
  */
 package org.apache.hc.core5.http2.impl.nio.bootstrap;
 
+import java.nio.channels.CancelledKeyException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 
@@ -48,7 +49,11 @@ public void setDependency(final Cancellable cancellable) {
         if (cancelled.get()) {
             final Cancellable dependency = dependencyRef.getAndSet(null);
             if (dependency != null) {
-                dependency.cancel();
+                try {
+                    dependency.cancel();
+                } catch (final CancelledKeyException ignore) {
+                    // Session already gone; cancellation is effectively complete.
+                }
             }
         }
     }
@@ -63,7 +68,11 @@ public boolean cancel() {
         if (cancelled.compareAndSet(false, true)) {
             final Cancellable dependency = dependencyRef.getAndSet(null);
             if (dependency != null) {
-                dependency.cancel();
+                try {
+                    dependency.cancel();
+                } catch (final CancelledKeyException ignore) {
+                    // Session already gone; treat as successfully cancelled.
+                }
             }
             return true;
         }