Stale check command for async connections

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

@@ -42,6 +42,7 @@
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
 
 import javax.net.ssl.SSLHandshakeException;
 import javax.net.ssl.SSLSession;
@@ -68,6 +69,7 @@
 import org.apache.hc.core5.http.nio.command.CommandSupport;
 import org.apache.hc.core5.http.nio.command.ExecutableCommand;
 import org.apache.hc.core5.http.nio.command.ShutdownCommand;
+import org.apache.hc.core5.http.nio.command.StaleCheckCommand;
 import org.apache.hc.core5.http.protocol.HttpCoreContext;
 import org.apache.hc.core5.http.protocol.HttpProcessor;
 import org.apache.hc.core5.http2.H2ConnectionException;
@@ -413,6 +415,11 @@ private int generateStreamId() {
         }
     }
 
+    void doStaleCheck(final Consumer<Boolean> callback) throws IOException {
+        callback.accept(ioSession.isOpen() &&
+            connState.compareTo(ConnectionHandshake.ACTIVE) == 0);
+    }
+
     public final void onConnect() throws HttpException, IOException {
         connState = ConnectionHandshake.ACTIVE;
         final RawFrame settingsFrame = frameFactory.createSettings(
@@ -656,6 +663,8 @@ private void processPendingCommands() throws IOException, HttpException {
                 if (!outputQueue.isEmpty()) {
                     return;
                 }
+            } else if (command instanceof StaleCheckCommand) {
+                doStaleCheck(((StaleCheckCommand) command).getCallback());
             }
         }
     }

httpcore5-h2/src/main/java/org/apache/hc/core5/http2/nio/pool/H2ConnPool.java

@@ -39,9 +39,8 @@
 import org.apache.hc.core5.http.URIScheme;
 import org.apache.hc.core5.http.impl.DefaultAddressResolver;
 import org.apache.hc.core5.http.nio.command.ShutdownCommand;
+import org.apache.hc.core5.http.nio.command.StaleCheckCommand;
 import org.apache.hc.core5.http.nio.ssl.TlsStrategy;
-import org.apache.hc.core5.http2.nio.command.PingCommand;
-import org.apache.hc.core5.http2.nio.support.BasicPingHandler;
 import org.apache.hc.core5.io.CloseMode;
 import org.apache.hc.core5.reactor.AbstractIOSessionPool;
 import org.apache.hc.core5.reactor.Command;
@@ -146,11 +145,7 @@ protected void validateSession(
                 final long lastAccessTime = Math.min(ioSession.getLastReadTime(), ioSession.getLastWriteTime());
                 final long deadline = lastAccessTime + timeValue.toMilliseconds();
                 if (deadline <= System.currentTimeMillis()) {
-                    final Timeout socketTimeoutMillis = ioSession.getSocketTimeout();
-                    ioSession.enqueue(new PingCommand(new BasicPingHandler(result -> {
-                        ioSession.setSocketTimeout(socketTimeoutMillis);
-                        callback.execute(result);
-                    })), Command.Priority.NORMAL);
+                    ioSession.enqueue(new StaleCheckCommand(callback::execute), Command.Priority.NORMAL);
                     return;
                 }
             }

httpcore5/src/main/java/org/apache/hc/core5/http/impl/nio/AbstractHttp1StreamDuplexer.java

@@ -36,6 +36,7 @@
 import java.util.List;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Consumer;
 
 import javax.net.ssl.SSLHandshakeException;
 import javax.net.ssl.SSLSession;
@@ -68,6 +69,7 @@
 import org.apache.hc.core5.http.nio.command.CommandSupport;
 import org.apache.hc.core5.http.nio.command.RequestExecutionCommand;
 import org.apache.hc.core5.http.nio.command.ShutdownCommand;
+import org.apache.hc.core5.http.nio.command.StaleCheckCommand;
 import org.apache.hc.core5.io.CloseMode;
 import org.apache.hc.core5.io.SocketTimeoutExceptionFactory;
 import org.apache.hc.core5.reactor.Command;
@@ -244,6 +246,8 @@ private void processCommands() throws HttpException, IOException {
                     execute((RequestExecutionCommand) command);
                     return;
                 }
+            } else if (command instanceof StaleCheckCommand) {
+                doStaleCheck(((StaleCheckCommand) command).getCallback());
             } else {
                 throw new HttpException("Unexpected command: " + command.getClass());
             }
@@ -428,6 +432,11 @@ void requestShutdown(final CloseMode closeMode) {
         ioSession.setEvent(SelectionKey.OP_WRITE);
     }
 
+    void doStaleCheck(final Consumer<Boolean> callback) throws IOException {
+        callback.accept(ioSession.isOpen() &&
+                connState.compareTo(ConnectionState.ACTIVE) == 0);
+    }
+
     void commitMessageHead(
             final OutgoingMessage messageHead,
             final boolean endStream,

httpcore5/src/main/java/org/apache/hc/core5/http/nio/command/StaleCheckCommand.java

@@ -0,0 +1,62 @@
+/*
+ * ====================================================================
+ * 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.http.nio.command;
+
+import java.util.function.Consumer;
+
+import org.apache.hc.core5.annotation.Internal;
+import org.apache.hc.core5.reactor.Command;
+import org.apache.hc.core5.util.Args;
+
+/**
+ * Stale check command. The {@code callback} will be invoked after the client's event loop has finished processing
+ * all pending reads on the connection. If the connection is still active at that point, the callback will be completed
+ * with {@code true}. In any other event, the callback will be cancelled, failed, or completed with {@code false}.
+ *
+ * @since 5.4
+ */
+@Internal
+public final class StaleCheckCommand implements Command {
+
+    private final Consumer<Boolean> callback;
+
+    public StaleCheckCommand(final Consumer<Boolean> callback) {
+        this.callback = Args.notNull(callback, "Callback");
+    }
+
+    public Consumer<Boolean> getCallback() {
+        return callback;
+    }
+
+    @Override
+    public boolean cancel() {
+        callback.accept(false);
+        return true;
+    }
+
+}