HTTPClient: add idle timeout and change timeout signatures

This adds an idle timeout based off PR #20.
In addition this changes the timeout TimeUnit to be the second argument (similar to how java typically orders the unit argument)

Author: jentfoo

client/src/main/java/org/threadly/litesockets/client/http/HTTPClient.java

@@ -5,6 +5,7 @@
 import java.net.URL;
 import java.nio.ByteBuffer;
 import java.util.ArrayDeque;
+import java.util.Iterator;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentLinkedQueue;
@@ -42,6 +43,7 @@
 import org.threadly.litesockets.utils.SSLUtils;
 import org.threadly.util.AbstractService;
 import org.threadly.util.Clock;
+import org.threadly.util.Pair;
 
 /**
  * <p>This is a HTTPClient for doing many simple HTTPRequests.  Every request will be make a new connection and requests
@@ -51,20 +53,23 @@
 public class HTTPClient extends AbstractService {
   public static final int DEFAULT_CONCURRENT = 2;
   public static final int DEFAULT_TIMEOUT = 15000;
+  public static final int DEFAULT_MAX_IDLE = 45000;
   public static final int MAX_HTTP_RESPONSE = 1048576;  //1MB
 
   private final int maxResponseSize;
   private final SubmitterScheduler ssi;
   private final SocketExecuter sei;
   private final ConcurrentLinkedQueue<HTTPRequestWrapper> queue = new ConcurrentLinkedQueue<>();
   private final ConcurrentHashMap<TCPClient, HTTPRequestWrapper> inProcess = new ConcurrentHashMap<>();
-  private final ConcurrentHashMap<HTTPAddress, ArrayDeque<TCPClient>> sockets = new ConcurrentHashMap<>();
+  private final ConcurrentHashMap<HTTPAddress, ArrayDeque<Pair<Long,TCPClient>>> sockets = new ConcurrentHashMap<>();
   private final CopyOnWriteArraySet<TCPClient> tcpClients = new CopyOnWriteArraySet<>();
   private final MainClientProcessor mcp = new MainClientProcessor();
   private final RunSocket runSocketTask;
   private final int maxConcurrent;
-  private volatile int defaultTimeoutMS = HTTPRequest.DEFAULT_TIMEOUT_MS;
+  private volatile Runnable checkIdle = null;
+  private volatile long defaultTimeoutMS = HTTPRequest.DEFAULT_TIMEOUT_MS;
   private volatile SSLContext sslContext = SSLUtils.OPEN_SSL_CTX;
+  private volatile long maxIdleTime = DEFAULT_MAX_IDLE;
 
   private NoThreadSocketExecuter ntse = null;
   private SingleThreadScheduler sts = null;
@@ -96,7 +101,6 @@ public HTTPClient(int maxConcurrent, int maxResponseSize) {
     runSocketTask = new RunSocket(ssi);
   }
 
-
   /**
    * <p>This constructor will let you set the max Concurrent Requests and max Response Size
    * as well as your own {@link SocketExecuter} as the thread pool to use.</p> 
@@ -168,8 +172,34 @@ public void closeAllClients() {
    * 
    * @param timeout time in milliseconds to wait for HTTPRequests to finish.
    */
-  public void setTimeout(TimeUnit unit, int timeout) {
-    this.defaultTimeoutMS = (int)Math.min(Math.max(unit.toMillis(timeout),HTTPRequest.MIN_TIMEOUT_MS), HTTPRequest.MAX_TIMEOUT_MS);
+  public void setTimeout(long timeout, TimeUnit unit) {
+    this.defaultTimeoutMS = Math.min(Math.max(unit.toMillis(timeout),HTTPRequest.MIN_TIMEOUT_MS), HTTPRequest.MAX_TIMEOUT_MS);
+  }
+  
+  public long getMaxIdleTimeout() {
+    return this.maxIdleTime;
+  }
+
+  /**
+   * Sets the max amount of time we will hold onto idle connections.  A 0 means we close connections when done, less
+   * than zero means we will never expire connections.
+   * 
+   * @param it the time in milliseconds to wait before timing out a connection.
+   */
+  public void setMaxIdleTimeout(long it, TimeUnit unit) {
+    this.maxIdleTime = unit.toMillis(it);
+    if(this.maxIdleTime > 0) {
+      this.checkIdle = new Runnable() {
+        @Override
+        public void run() {
+          if(checkIdle == this) {
+            checkIdleSockets();
+            ssi.schedule(this, Math.max(100, maxIdleTime/2));
+          }
+        }
+      };
+      this.ssi.schedule(checkIdle, Math.max(100, maxIdleTime/2));
+    }
   }
 
   /**
@@ -260,11 +290,10 @@ public ListenableFuture<HTTPResponseData> requestAsync(final URL url) {
   public ListenableFuture<HTTPResponseData> requestAsync(final URL url, final HTTPRequestType rt, final ByteBuffer bb) {
     HTTPRequestBuilder hrb = new HTTPRequestBuilder(url);
     hrb.setRequestType(rt);
-    hrb.setTimeout(TimeUnit.MILLISECONDS, this.defaultTimeoutMS);
+    hrb.setTimeout(this.defaultTimeoutMS, TimeUnit.MILLISECONDS);
     return requestAsync(hrb.buildClientHTTPRequest());
   }
 
-
   /**
    * Sends an asynchronous HTTP request.
    * 
@@ -319,6 +348,7 @@ protected void startupService() {
     if (ntse != null) {
       ntse.start();
     }
+    setMaxIdleTimeout(this.maxIdleTime, TimeUnit.MILLISECONDS);
   }
 
   @Override
@@ -338,18 +368,18 @@ protected void shutdownService() {
   }
 
   private TCPClient getTCPClient(final HTTPAddress ha) throws IOException {
-    ArrayDeque<TCPClient> ll = sockets.get(ha);
+    ArrayDeque<Pair<Long,TCPClient>> pl = sockets.get(ha);
     TCPClient tc = null;
-    if(ll != null) {
-      synchronized(ll) {
-        while(ll.size() > 0 && tc == null) {
-          if(ll.peek().isClosed()) {
-            ll.pop();
+    if(pl != null) {
+      synchronized(pl) {
+        while(pl.size() > 0 && tc == null) {
+          if(pl.peek().getRight().isClosed()) {
+            pl.pop();
           } else {
-            tc = ll.pop();
+            tc = pl.pop().getRight();
           }
         }
-        if(ll.size() == 0) {
+        if(pl.size() == 0) {
           sockets.remove(ha);
         }
       }
@@ -371,14 +401,35 @@ private TCPClient getTCPClient(final HTTPAddress ha) throws IOException {
   }
 
   private void addBackTCPClient(final HTTPAddress ha, final TCPClient client) {
+    if(maxIdleTime == 0) {
+      client.close();
+      return;
+    }
     if(!client.isClosed()) {
-      ArrayDeque<TCPClient> ll = sockets.get(ha);  
+      ArrayDeque<Pair<Long,TCPClient>> ll = sockets.get(ha);  
       if(ll == null) {
         sockets.put(ha, new ArrayDeque<>(8));
         ll = sockets.get(ha);
       }
       synchronized(ll) {
-        ll.add(client);
+        ll.add(new Pair<>(Clock.lastKnownForwardProgressingMillis(), client));
+      }
+    }
+  }
+  
+  private void checkIdleSockets() {
+    if(maxIdleTime > 0) {
+      for(ArrayDeque<Pair<Long,TCPClient>> adq: sockets.values()) {
+        synchronized(adq) {
+          Iterator<Pair<Long,TCPClient>> iter = adq.iterator();
+          while(iter.hasNext()) {
+            Pair<Long,TCPClient> c = iter.next();
+            if(Clock.lastKnownForwardProgressingMillis() - c.getLeft() > maxIdleTime) {
+              iter.remove();
+              c.getRight().close();
+            }
+          }
+        }
       }
     }
   }
@@ -417,7 +468,7 @@ public void onClose(Client client) {
       if(hrw != null) {
         boolean wasProcessing = hrw.hrp.isProcessing();
         hrw.hrp.connectionClosed();
-        if(!hrw.slf.isDone() && !wasProcessing) {
+        if(! hrw.slf.isDone() && ! wasProcessing) {
           hrw.client = null;
           process(hrw);  
         } else {
@@ -554,3 +605,4 @@ public String toString() {
     }
   }
 }
+

client/src/test/java/org/threadly/litesockets/client/http/HTTPClientTests.java

@@ -5,7 +5,6 @@
 import static org.junit.Assert.fail;
 
 import java.io.IOException;
-import java.net.MalformedURLException;
 import java.net.URL;
 import java.nio.ByteBuffer;
 import java.util.HashMap;
@@ -25,8 +24,6 @@
 import org.threadly.litesockets.SocketExecuter;
 import org.threadly.litesockets.TCPServer;
 import org.threadly.litesockets.ThreadedSocketExecuter;
-import org.threadly.litesockets.buffers.MergedByteBuffers;
-import org.threadly.litesockets.buffers.ReuseableMergedByteBuffers;
 import org.threadly.litesockets.client.http.HTTPClient.HTTPResponseData;
 import org.threadly.litesockets.client.http.HTTPStreamClient.HTTPStreamReader;
 import org.threadly.litesockets.protocols.http.request.ClientHTTPRequest;
@@ -300,7 +297,7 @@ public void closeBeforeLength() throws IOException, HTTPParsingException {
     fakeServer = new TestHTTPServer(port, RESPONSE_HUGE, CONTENT, false, true);
     final HTTPRequestBuilder hrb = new HTTPRequestBuilder(new URL("http://localhost:"+port));
     hrb.setBody(IOUtils.EMPTY_BYTEBUFFER);
-    hrb.setTimeout(TimeUnit.MILLISECONDS, 10000);
+    hrb.setTimeout(10000, TimeUnit.MILLISECONDS);
     final HTTPClient httpClient = new HTTPClient();
     httpClient.start();
     try{
@@ -319,7 +316,7 @@ public void timeoutRequest() throws IOException, HTTPParsingException {
     server.start();
     final HTTPRequestBuilder hrb = new HTTPRequestBuilder(new URL("http://localhost:"+port));
     hrb.setBody(IOUtils.EMPTY_BYTEBUFFER);
-    hrb.setTimeout(TimeUnit.MILLISECONDS, 500);
+    hrb.setTimeout(500, TimeUnit.MILLISECONDS);
     final HTTPClient httpClient = new HTTPClient();
     httpClient.start();
     try{
@@ -336,7 +333,7 @@ public void expireRequest() throws IOException, HTTPParsingException {
     fakeServer = new TestHTTPServer(port, RESPONSE_CL, "", false, false);
     final HTTPRequestBuilder hrb = new HTTPRequestBuilder(new URL("http://localhost:"+port));
     hrb.setBody(IOUtils.EMPTY_BYTEBUFFER);
-    hrb.setTimeout(TimeUnit.MILLISECONDS, 30);
+    hrb.setTimeout(30, TimeUnit.MILLISECONDS);
     final HTTPClient httpClient = new HTTPClient();
     httpClient.start();
     long start = Clock.accurateForwardProgressingMillis();
@@ -430,7 +427,7 @@ public void accept(Client c) {
       }});
     server.start();
     final HTTPRequestBuilder hrb = new HTTPRequestBuilder(new URL("http://localhost:"+port));
-    hrb.setTimeout(TimeUnit.MILLISECONDS, 500);
+    hrb.setTimeout(500, TimeUnit.MILLISECONDS);
     hrb.setBody(IOUtils.EMPTY_BYTEBUFFER);
     final HTTPClient httpClient = new HTTPClient();
     httpClient.start();

protocol/src/main/java/org/threadly/litesockets/protocols/http/request/HTTPRequestBuilder.java

@@ -240,7 +240,7 @@ public HTTPRequestBuilder setBody(final String str, Charset cs) {
     return setBody(ByteBuffer.wrap(str.getBytes(cs)));
   }
 
-  public HTTPRequestBuilder setTimeout(TimeUnit unit, int size) {
+  public HTTPRequestBuilder setTimeout(long size, TimeUnit unit) {
     this.timeoutMS = (int)Math.min(Math.max(unit.toMillis(size),HTTPRequest.MIN_TIMEOUT_MS), HTTPRequest.MAX_TIMEOUT_MS);
     return this;
   }