HTTPCORE-761: support ExtendedSocketOption in socket

httpcore5/src/main/java/org/apache/hc/core5/reactor/IOReactorConfig.java

@@ -57,6 +57,9 @@ public final class IOReactorConfig {
     private final int sndBufSize;
     private final int rcvBufSize;
     private final int backlogSize;
+    private final int tcpKeepIdle;
+    private final int tcpKeepInterval;
+    private final int tcpKeepCount;
     private final SocketAddress socksProxyAddress;
     private final String socksProxyUsername;
     private final String socksProxyPassword;
@@ -73,6 +76,9 @@ public final class IOReactorConfig {
             final int sndBufSize,
             final int rcvBufSize,
             final int backlogSize,
+            final int tcpKeepIdle,
+            final int tcpKeepInterval,
+            final int tcpKeepCount,
             final SocketAddress socksProxyAddress,
             final String socksProxyUsername,
             final String socksProxyPassword) {
@@ -88,6 +94,9 @@ public final class IOReactorConfig {
         this.sndBufSize = sndBufSize;
         this.rcvBufSize = rcvBufSize;
         this.backlogSize = backlogSize;
+        this.tcpKeepIdle = tcpKeepIdle;
+        this.tcpKeepInterval = tcpKeepInterval;
+        this.tcpKeepCount = tcpKeepCount;
         this.socksProxyAddress = socksProxyAddress;
         this.socksProxyUsername = socksProxyUsername;
         this.socksProxyPassword = socksProxyPassword;
@@ -182,6 +191,33 @@ public int getBacklogSize() {
         return backlogSize;
     }
 
+    /**
+     * @see Builder#setTcpKeepIdle(int)
+     *
+     * @since 5.3
+     */
+    public int getTcpKeepIdle() {
+        return this.tcpKeepIdle;
+    }
+
+    /**
+     * @see Builder#setTcpKeepInterval(int)
+     *
+     * @since 5.3
+     */
+    public int getTcpKeepInterval() {
+        return this.tcpKeepInterval;
+    }
+
+    /**
+     * @see Builder#setTcpKeepCount(int)
+     *
+     * @since 5.3
+     */
+    public int getTcpKeepCount() {
+        return this.tcpKeepCount;
+    }
+
     /**
      * @see Builder#setSocksProxyAddress(SocketAddress)
      */
@@ -220,6 +256,9 @@ public static Builder copy(final IOReactorConfig config) {
             .setSndBufSize(config.getSndBufSize())
             .setRcvBufSize(config.getRcvBufSize())
             .setBacklogSize(config.getBacklogSize())
+            .setTcpKeepIdle(config.getTcpKeepIdle())
+            .setTcpKeepInterval(config.getTcpKeepInterval())
+            .setTcpKeepCount(config.getTcpKeepCount())
             .setSocksProxyAddress(config.getSocksProxyAddress())
             .setSocksProxyUsername(config.getSocksProxyUsername())
             .setSocksProxyPassword(config.getSocksProxyPassword());
@@ -265,6 +304,9 @@ public static void setDefaultMaxIOThreadCount(final int defaultMaxIOThreadCount)
         private int sndBufSize;
         private int rcvBufSize;
         private int backlogSize;
+        private int tcpKeepIdle;
+        private int tcpKeepInterval;
+        private int tcpKeepCount;
         private SocketAddress socksProxyAddress;
         private String socksProxyUsername;
         private String socksProxyPassword;
@@ -281,6 +323,9 @@ public static void setDefaultMaxIOThreadCount(final int defaultMaxIOThreadCount)
             this.sndBufSize = 0;
             this.rcvBufSize = 0;
             this.backlogSize = 0;
+            this.tcpKeepIdle = -1;
+            this.tcpKeepInterval = -1;
+            this.tcpKeepCount = -1;
             this.socksProxyAddress = null;
             this.socksProxyUsername = null;
             this.socksProxyPassword = null;
@@ -462,6 +507,37 @@ public Builder setBacklogSize(final int backlogSize) {
             return this;
         }
 
+        /**
+         * Determines the time (in seconds) the connection needs to remain idle before TCP starts
+         * sending keepalive probes.
+         *
+         * @since 5.3
+         */
+        public Builder setTcpKeepIdle(final int tcpKeepIdle) {
+            this.tcpKeepIdle = tcpKeepIdle;
+            return this;
+        }
+
+        /**
+         * Determines the time (in seconds) between individual keepalive probes.
+         *
+         * @since 5.3
+         */
+        public Builder setTcpKeepInterval(final int tcpKeepInterval) {
+            this.tcpKeepInterval = tcpKeepInterval;
+            return this;
+        }
+
+        /**
+         * Determines the maximum number of keepalive probes TCP should send before dropping the connection.
+         *
+         * @since 5.3
+         */
+        public Builder setTcpKeepCount(final int tcpKeepCount) {
+            this.tcpKeepCount = tcpKeepCount;
+            return this;
+        }
+
         /**
          * The address of the SOCKS proxy to use.
          */
@@ -497,6 +573,7 @@ public IOReactorConfig build() {
                     tcpNoDelay,
                     trafficClass,
                     sndBufSize, rcvBufSize, backlogSize,
+                    tcpKeepIdle, tcpKeepInterval, tcpKeepCount,
                     socksProxyAddress, socksProxyUsername, socksProxyPassword);
         }
 

httpcore5/src/main/java/org/apache/hc/core5/reactor/SingleCoreIOReactor.java

@@ -27,10 +27,22 @@
 
 package org.apache.hc.core5.reactor;
 
+import org.apache.hc.core5.annotation.Internal;
+import org.apache.hc.core5.concurrent.FutureCallback;
+import org.apache.hc.core5.function.Callback;
+import org.apache.hc.core5.function.Decorator;
+import org.apache.hc.core5.io.CloseMode;
+import org.apache.hc.core5.io.Closer;
+import org.apache.hc.core5.net.NamedEndpoint;
+import org.apache.hc.core5.util.Args;
+import org.apache.hc.core5.util.Asserts;
+import org.apache.hc.core5.util.Timeout;
+
 import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.net.Socket;
 import java.net.SocketAddress;
+import java.net.SocketOption;
 import java.net.UnknownHostException;
 import java.nio.channels.CancelledKeyException;
 import java.nio.channels.ClosedChannelException;
@@ -45,18 +57,14 @@
 import java.util.concurrent.Future;
 import java.util.concurrent.atomic.AtomicBoolean;
 
-import org.apache.hc.core5.concurrent.FutureCallback;
-import org.apache.hc.core5.function.Callback;
-import org.apache.hc.core5.function.Decorator;
-import org.apache.hc.core5.io.CloseMode;
-import org.apache.hc.core5.io.Closer;
-import org.apache.hc.core5.net.NamedEndpoint;
-import org.apache.hc.core5.util.Args;
-import org.apache.hc.core5.util.Asserts;
-import org.apache.hc.core5.util.Timeout;
+import static org.apache.hc.core5.util.ReflectionUtils.getExtendedSocketOptionOrNull;
 
-class SingleCoreIOReactor extends AbstractSingleCoreIOReactor implements ConnectionInitiator {
+@Internal
+public class SingleCoreIOReactor extends AbstractSingleCoreIOReactor implements ConnectionInitiator {
 
+    public static final String TCP_KEEPIDLE = "TCP_KEEPIDLE";
+    public static final String TCP_KEEPINTERVAL = "TCP_KEEPINTERVAL";
+    public static final String TCP_KEEPCOUNT = "TCP_KEEPCOUNT";
     private static final int MAX_CHANNEL_REQUESTS = 10000;
 
     private final IOEventHandlerFactory eventHandlerFactory;
@@ -190,7 +198,7 @@ private void processPendingChannels() throws IOException {
             final SocketChannel socketChannel = entry.channel;
             final Object attachment = entry.attachment;
             try {
-                prepareSocket(socketChannel.socket());
+                prepareSocket(socketChannel);
                 socketChannel.configureBlocking(false);
             } catch (final IOException ex) {
                 logException(ex);
@@ -265,7 +273,8 @@ public Future<IOSession> connect(
         return sessionRequest;
     }
 
-    private void prepareSocket(final Socket socket) throws IOException {
+    private void prepareSocket(final SocketChannel socketChannel) throws IOException {
+        final Socket socket = socketChannel.socket();
         socket.setTcpNoDelay(this.reactorConfig.isTcpNoDelay());
         socket.setKeepAlive(this.reactorConfig.isSoKeepAlive());
         if (this.reactorConfig.getSndBufSize() > 0) {
@@ -281,6 +290,27 @@ private void prepareSocket(final Socket socket) throws IOException {
         if (linger >= 0) {
             socket.setSoLinger(true, linger);
         }
+        if (this.reactorConfig.getTcpKeepIdle() > 0) {
+            setExtendedSocketOption(socketChannel, TCP_KEEPIDLE, this.reactorConfig.getTcpKeepIdle());
+        }
+        if (this.reactorConfig.getTcpKeepInterval() > 0) {
+            setExtendedSocketOption(socketChannel, TCP_KEEPINTERVAL, this.reactorConfig.getTcpKeepInterval());
+        }
+        if (this.reactorConfig.getTcpKeepInterval() > 0) {
+            setExtendedSocketOption(socketChannel, TCP_KEEPCOUNT, this.reactorConfig.getTcpKeepCount());
+        }
+    }
+
+    /**
+     * @since 5.3
+     */
+    <T> void setExtendedSocketOption(final SocketChannel socketChannel,
+                                             final String optionName, final T value) throws IOException {
+        final SocketOption<T> socketOption = getExtendedSocketOptionOrNull(optionName);
+        if (socketOption == null) {
+            throw new UnsupportedOperationException(optionName + " is not supported in the current jdk");
+        }
+        socketChannel.setOption(socketOption, value);
     }
 
     private void validateAddress(final SocketAddress address) throws UnknownHostException {
@@ -315,7 +345,7 @@ private void processPendingConnectionRequests() {
 
     private void processConnectionRequest(final SocketChannel socketChannel, final IOSessionRequest sessionRequest) throws IOException {
         socketChannel.configureBlocking(false);
-        prepareSocket(socketChannel.socket());
+        prepareSocket(socketChannel);
 
         validateAddress(sessionRequest.localAddress);
         if (sessionRequest.localAddress != null) {

httpcore5/src/main/java/org/apache/hc/core5/util/ReflectionUtils.java

@@ -27,7 +27,9 @@
 
 package org.apache.hc.core5.util;
 
+import java.lang.reflect.Field;
 import java.lang.reflect.Method;
+import java.net.SocketOption;
 import java.security.AccessController;
 import java.security.PrivilegedAction;
 
@@ -81,4 +83,16 @@ public static int determineJRELevel() {
         return 7;
     }
 
+    /**
+     * @since 5.3
+     */
+    public static <T> SocketOption<T> getExtendedSocketOptionOrNull(final String fieldName) {
+        try {
+            final Class<?> extendedSocketOptionsClass = Class.forName("jdk.net.ExtendedSocketOptions");
+            final Field field = extendedSocketOptionsClass.getField(fieldName);
+            return (SocketOption)field.get((Object)null);
+        } catch (final Exception ignore) {
+            return null;
+        }
+    }
 }

httpcore5/src/test/java/org/apache/hc/core5/reactor/IOReactorConfigTest.java

@@ -48,6 +48,9 @@ public void testCustomIOReactorConfig() throws Exception {
                 .setSndBufSize(32767)
                 .setRcvBufSize(8192)
                 .setBacklogSize(5)
+                .setTcpKeepIdle(100)
+                .setTcpKeepInterval(12)
+                .setTcpKeepCount(4)
                 .setSocksProxyAddress(new InetSocketAddress(8888))
                 .setSocksProxyUsername("socksProxyUsername")
                 .setSocksProxyPassword("socksProxyPassword")
@@ -64,6 +67,9 @@ public void testCustomIOReactorConfig() throws Exception {
         Assertions.assertEquals(32767, reactorConfig.getSndBufSize());
         Assertions.assertEquals(8192, reactorConfig.getRcvBufSize());
         Assertions.assertEquals(5, reactorConfig.getBacklogSize());
+        Assertions.assertEquals(100, reactorConfig.getTcpKeepIdle());
+        Assertions.assertEquals(12, reactorConfig.getTcpKeepInterval());
+        Assertions.assertEquals(4, reactorConfig.getTcpKeepCount());
         Assertions.assertEquals(new InetSocketAddress(8888), reactorConfig.getSocksProxyAddress());
         Assertions.assertEquals("socksProxyUsername", reactorConfig.getSocksProxyUsername());
         Assertions.assertEquals("socksProxyPassword", reactorConfig.getSocksProxyPassword());

httpcore5/src/test/java/org/apache/hc/core5/reactor/TestSingleCoreIOReactor.java

@@ -0,0 +1,66 @@
+/*
+ * ====================================================================
+ * 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.reactor;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.net.SocketOption;
+import java.nio.channels.SocketChannel;
+
+import static org.apache.hc.core5.reactor.SingleCoreIOReactor.TCP_KEEPCOUNT;
+import static org.apache.hc.core5.reactor.SingleCoreIOReactor.TCP_KEEPIDLE;
+import static org.apache.hc.core5.reactor.SingleCoreIOReactor.TCP_KEEPINTERVAL;
+import static org.apache.hc.core5.util.ReflectionUtils.determineJRELevel;
+import static org.apache.hc.core5.util.ReflectionUtils.getExtendedSocketOptionOrNull;
+import static org.mockito.Mockito.mock;
+
+public class TestSingleCoreIOReactor {
+    @Test
+    public void testSetExtendedSocketOption() throws IOException {
+        final SingleCoreIOReactor reactor = new SingleCoreIOReactor(null, mock(IOEventHandlerFactory.class), IOReactorConfig.DEFAULT, null, null, null);
+        final SocketChannel socketChannel = SocketChannel.open();
+        final SocketOption<Integer> tcpKeepIdle;
+        final SocketOption<Integer> tcpKeepInterval;
+        final SocketOption<Integer> tcpKeepCount;
+        // Partial versions of jdk1.8 contain TCP_KEEPIDLE, TCP_KEEPINTERVAL, TCP_KEEPCOUNT.
+        if (determineJRELevel() > 8) {
+            reactor.setExtendedSocketOption(socketChannel, TCP_KEEPIDLE, 100);
+            reactor.setExtendedSocketOption(socketChannel, TCP_KEEPINTERVAL, 10);
+            reactor.setExtendedSocketOption(socketChannel, TCP_KEEPCOUNT, 10);
+
+            tcpKeepIdle = getExtendedSocketOptionOrNull(TCP_KEEPIDLE);
+            tcpKeepInterval = getExtendedSocketOptionOrNull(TCP_KEEPINTERVAL);
+            tcpKeepCount = getExtendedSocketOptionOrNull(TCP_KEEPCOUNT);
+            Assertions.assertEquals(100, socketChannel.getOption(tcpKeepIdle));
+            Assertions.assertEquals(10, socketChannel.getOption(tcpKeepInterval));
+            Assertions.assertEquals(10, socketChannel.getOption(tcpKeepCount));
+        }
+    }
+}