Stale check command for async connections

Author: ok2c

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

@@ -41,11 +41,10 @@
 import java.util.concurrent.ConcurrentLinkedDeque;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.atomic.AtomicInteger;
-
 import javax.net.ssl.SSLHandshakeException;
 import javax.net.ssl.SSLSession;
-
 import org.apache.hc.core5.concurrent.CancellableDependency;
+import org.apache.hc.core5.concurrent.FutureCallback;
 import org.apache.hc.core5.http.ConnectionClosedException;
 import org.apache.hc.core5.http.EndpointDetails;
 import org.apache.hc.core5.http.Header;
@@ -65,6 +64,7 @@
 import org.apache.hc.core5.http.nio.HandlerFactory;
 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;
@@ -410,6 +410,23 @@ private int generateStreamId() {
         }
     }
 
+    void doStalecheck(final FutureCallback<Boolean> callback) throws IOException {
+        if (!ioSession.isOpen() || connState.compareTo(ConnectionHandshake.ACTIVE) > 0) {
+            callback.completed(false);
+        }
+        final ByteBuffer buffer = ByteBuffer.allocate(0);
+        final int bytesRead = ioSession.channel().read(buffer);
+        if (bytesRead == -1) {
+            if (connState == ConnectionHandshake.ACTIVE) {
+                connState = ConnectionHandshake.GRACEFUL_SHUTDOWN;
+            }
+            requestSessionOutput();
+            callback.completed(false);
+        } else {
+            callback.completed(true);
+        }
+    }
+
     public final void onConnect() throws HttpException, IOException {
         connState = ConnectionHandshake.ACTIVE;
         final RawFrame settingsFrame = frameFactory.createSettings(
@@ -647,6 +664,8 @@ private void processPendingCommands() throws IOException, HttpException {
                 if (!outputQueue.isEmpty()) {
                     return;
                 }
+            } else if (command instanceof StaleCheckCommand) {
+                doStalecheck(((StaleCheckCommand) command).getCallback());
             }
         }
     }

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

@@ -40,6 +40,7 @@
 import javax.net.ssl.SSLHandshakeException;
 import javax.net.ssl.SSLSession;
 
+import org.apache.hc.core5.concurrent.FutureCallback;
 import org.apache.hc.core5.http.ConnectionClosedException;
 import org.apache.hc.core5.http.ContentLengthStrategy;
 import org.apache.hc.core5.http.EndpointDetails;
@@ -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());
             }
@@ -429,6 +433,20 @@ void requestShutdown(final CloseMode closeMode) {
         ioSession.setEvent(SelectionKey.OP_WRITE);
     }
 
+    void doStalecheck(final FutureCallback<Boolean> callback) throws IOException {
+        if (!ioSession.isOpen() || connState.compareTo(ConnectionState.ACTIVE) > 0) {
+            callback.completed(false);
+        }
+        final ByteBuffer buffer = ByteBuffer.allocate(0);
+        final int bytesRead = ioSession.channel().read(buffer);
+        if (bytesRead == -1) {
+            requestShutdown(CloseMode.GRACEFUL);
+            callback.completed(false);
+        } else {
+            callback.completed(true);
+        }
+    }
+
     void commitMessageHead(
             final OutgoingMessage messageHead,
             final boolean endStream,

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

@@ -0,0 +1,58 @@
+/*
+ * ====================================================================
+ * 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 org.apache.hc.core5.annotation.Internal;
+import org.apache.hc.core5.concurrent.FutureCallback;
+import org.apache.hc.core5.reactor.Command;
+import org.apache.hc.core5.util.Args;
+
+/**
+ * Stale check command.
+ *
+ * @since 5.4
+ */
+@Internal
+public final class StaleCheckCommand implements Command {
+
+    private final FutureCallback<Boolean> callback;
+
+    public StaleCheckCommand(final FutureCallback<Boolean> callback) {
+        this.callback = Args.notNull(callback, "Callback");
+    }
+
+    public FutureCallback<Boolean> getCallback() {
+        return callback;
+    }
+
+    @Override
+    public boolean cancel() {
+        return true;
+    }
+
+}