HttpClient: Allow queue limit and change request exceptions

This commit includes two changes:
1) Allows you to set a queue limit for requests

It's still recommended the queue size to be at least as large as the number of concurrent requests due to the fact that all requests must pass through the queue.

This limit however can aleviate conditions where we otherwise would need to timeout by being in queue so long.  Instead we can now fail fast when the client is not making progress

2) This changes the exceptions which may be thrown from `request` to no longer only be `HTTPParsingException`

This is to make the behavior uniform with asyncRequest.  AsyncRequest may throw a CancellationException, a RejectedExecutionException, or other cases which are not only `HTTPParsingException`.  Rather than changing the async request to always return this exception, I think it's better to let the specific exception types bubble up.  Alternatively we could create `http` exceptions for each condition we want to report, however I don't believe it's worth the per-request overhead to map these exception types.

Author: jentfoo

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

@@ -6,10 +6,14 @@
 import java.nio.ByteBuffer;
 import java.util.ArrayDeque;
 import java.util.Iterator;
+import java.util.Queue;
+import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.CopyOnWriteArraySet;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.TimeUnit;
 
 import javax.net.ssl.SSLContext;
@@ -18,6 +22,7 @@
 import org.threadly.concurrent.ReschedulingOperation;
 import org.threadly.concurrent.SingleThreadScheduler;
 import org.threadly.concurrent.SubmitterScheduler;
+import org.threadly.concurrent.future.FutureUtils;
 import org.threadly.concurrent.future.ListenableFuture;
 import org.threadly.concurrent.future.SettableListenableFuture;
 import org.threadly.litesockets.Client;
@@ -60,7 +65,7 @@ public class HTTPClient extends AbstractService {
   private final int maxResponseSize;
   private final SubmitterScheduler ssi;
   private final SocketExecuter sei;
-  private final ConcurrentLinkedQueue<HTTPRequestWrapper> queue = new ConcurrentLinkedQueue<>();
+  private final Queue<HTTPRequestWrapper> queue;
   private final ConcurrentHashMap<TCPClient, HTTPRequestWrapper> inProcess = new ConcurrentHashMap<>();
   private final ConcurrentHashMap<HTTPAddress, ArrayDeque<Pair<Long,TCPClient>>> sockets = new ConcurrentHashMap<>();
   private final CopyOnWriteArraySet<TCPClient> tcpClients = new CopyOnWriteArraySet<>();
@@ -93,12 +98,34 @@ public HTTPClient() {
    * @param maxResponseSize the maximum responseSize clients are allowed to send.
    */
   public HTTPClient(int maxConcurrent, int maxResponseSize) {
+    this(maxConcurrent, maxResponseSize, -1);
+  }
+
+  /**
+   * <p>This constructor will let you set the max Concurrent Requests and max Response Size but will still 
+   * create its own {@link SingleThreadScheduler} to use as a threadpool.</p>
+   *  
+   * <p>The maximum queue length parameter can help improved conditions where you want to fail fast 
+   * if the downstream service is fully consumed.  Since all requests will start in the queue we 
+   * recommend to either provide a {@code 0} to leave the queue unbounded, or to set to at least 
+   * the {@code maxConcurrent} value.</p>
+   * 
+   * @param maxConcurrent maximum number of requests to run simultaneously. 
+   * @param maxResponseSize the maximum responseSize clients are allowed to send.
+   * @param maxQueueSize Maximum queue size, {@code <= 0} to leave unbounded.  Recommended to be >= {@code maxConcurrent}
+   */
+  public HTTPClient(int maxConcurrent, int maxResponseSize, int maxQueueSize) {
     this.maxConcurrent = maxConcurrent;
     this.maxResponseSize = maxResponseSize;
     sts = new SingleThreadScheduler();
     this.ssi = sts;
     ntse = new NoThreadSocketExecuter();
     sei = ntse;
+    if (maxQueueSize < 1 || maxQueueSize == Integer.MAX_VALUE) {
+      queue = new ConcurrentLinkedQueue<>();
+    } else {
+      queue = new ArrayBlockingQueue<>(maxQueueSize);
+    }
     runSocketTask = new RunSocket(ssi);
   }
 
@@ -111,10 +138,33 @@ public HTTPClient(int maxConcurrent, int maxResponseSize) {
    * @param sei the SocketExecuter to use with these HTTPClients.
    */
   public HTTPClient(int maxConcurrent, int maxResponseSize, SocketExecuter sei) {
+    this(maxConcurrent, maxResponseSize, sei, -1);
+  }
+
+  /**
+   * <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>
+   * 
+   * <p>The maximum queue length parameter can help improved conditions where you want to fail fast 
+   * if the downstream service is fully consumed.  Since all requests will start in the queue we 
+   * recommend to either provide a {@code 0} to leave the queue unbounded, or to set to at least 
+   * the {@code maxConcurrent} value.</p>
+   * 
+   * @param maxConcurrent maximum number of requests to run simultaneously. 
+   * @param maxResponseSize the maximum responseSize clients are allowed to send.
+   * @param sei the SocketExecuter to use with these HTTPClients.
+   * @param maxQueueSize Maximum queue size, {@code <= 0} to be unbounded.  Recommended to be {@code >= maxConcurrent}
+   */
+  public HTTPClient(int maxConcurrent, int maxResponseSize, SocketExecuter sei, int maxQueueSize) {
     this.maxConcurrent = maxConcurrent;
     this.maxResponseSize = maxResponseSize;
     this.ssi = sei.getThreadScheduler();
     this.sei = sei;
+    if (maxQueueSize < 1 || maxQueueSize == Integer.MAX_VALUE) {
+      queue = new ConcurrentLinkedQueue<>();
+    } else {
+      queue = new ArrayBlockingQueue<>(maxQueueSize);
+    }
     runSocketTask = new RunSocket(ssi);
   }
 
@@ -232,19 +282,7 @@ public HTTPResponseData request(final URL url) throws HTTPParsingException {
    * @throws HTTPParsingException is thrown if the server sends back protocol or a response that is larger then allowed.
    */
   public HTTPResponseData request(final URL url, final HTTPRequestMethod rm, final ByteBuffer bb) throws HTTPParsingException {
-    HTTPResponseData hr = null;
-    try {
-      hr = requestAsync(url, rm, bb).get();
-    } catch (InterruptedException e) {
-      Thread.currentThread().interrupt();
-    } catch (Exception e) {
-      if(e.getCause() instanceof HTTPParsingException) {
-        throw (HTTPParsingException)e.getCause();
-      } else {
-        throw new HTTPParsingException(e);
-      }
-    }
-    return hr;
+    return extractAsyncResponse(requestAsync(url, rm, bb));
   }
 
   /**
@@ -255,21 +293,27 @@ public HTTPResponseData request(final URL url, final HTTPRequestMethod rm, final
    * @throws HTTPParsingException is thrown if the server sends back protocol or a response that is larger then allowed.
    */
   public HTTPResponseData request(final ClientHTTPRequest request) throws HTTPParsingException {
-    HTTPResponseData hr = null;
+    return extractAsyncResponse(requestAsync(request));
+  }
+  
+  protected HTTPResponseData extractAsyncResponse(ListenableFuture<HTTPResponseData> lf) throws HTTPParsingException {
     try {
-      hr = requestAsync(request).get();
+      return lf.get();
     } catch (InterruptedException e) {
       Thread.currentThread().interrupt();
-    } catch (Exception e) {
-      if(e.getCause() instanceof HTTPParsingException) {
+      lf.cancel(true);
+      throw new RuntimeException("Request interrupted", e);
+    } catch (CancellationException e) {
+      throw e;
+    } catch (ExecutionException e) {
+      if (e.getCause() instanceof HTTPParsingException) {
         throw (HTTPParsingException)e.getCause();
-      } else if(e instanceof CancellationException) {
-        throw new HTTPParsingException("HTTP Timeout!", e);
+      } else if (e.getCause() instanceof RejectedExecutionException) {
+        throw (RejectedExecutionException)e.getCause();
       } else {
         throw new HTTPParsingException(e);
       }
     }
-    return hr;
   }
 
   /**
@@ -312,7 +356,9 @@ public ListenableFuture<HTTPResponseData> requestAsync(final URL url, final HTTP
    */
   public ListenableFuture<HTTPResponseData> requestAsync(final ClientHTTPRequest request) {
     HTTPRequestWrapper hrw = new HTTPRequestWrapper(request);
-    queue.add(hrw);
+    if (! queue.offer(hrw)) {
+      return FutureUtils.immediateFailureFuture(new RejectedExecutionException("Request queue full"));
+    }
     if(ntse != null) {
       ntse.wakeup();
       runSocketTask.signalToRun();

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

@@ -7,7 +7,9 @@
 import java.io.IOException;
 import java.net.URL;
 import java.util.HashMap;
+import java.util.concurrent.CancellationException;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -309,8 +311,8 @@ public void timeoutRequest() throws IOException, HTTPParsingException {
     try{
       httpClient.request(hrb.buildClientHTTPRequest());
       fail();
-    } catch(HTTPParsingException e) {
-      assertEquals("HTTP Timeout!", e.getMessage());
+    } catch(CancellationException e) {
+      assertEquals("Request timed out at point: SendingRequest", e.getMessage());
       // below conditions may be slightly async due to future getting a result before listeners are invoked
       new TestCondition(() -> httpClient.getRequestQueueSize() == 0).blockTillTrue(1_000);
       new TestCondition(() -> httpClient.getInProgressSize() == 0).blockTillTrue(1_000);
@@ -335,13 +337,37 @@ protected void processQueue() {
     try{
       httpClient.request(hrb.buildClientHTTPRequest());
       fail();
-    } catch(HTTPParsingException e) {
-      assertEquals("HTTP Timeout!", e.getMessage());
+    } catch(CancellationException e) {
+      assertEquals("Request timed out at point: Queued", e.getMessage());
       // below conditions may be slightly async due to future getting a result before listeners are invoked
       new TestCondition(() -> httpClient.getRequestQueueSize() == 0).blockTillTrue(1_000);
       new TestCondition(() -> httpClient.getInProgressSize() == 0).blockTillTrue(1_000);
     }
   }
+  
+  @Test
+  public void queueLimitTest() throws IOException, HTTPParsingException {
+    int port = PortUtils.findTCPPort();
+    TCPServer server = SEI.createTCPServer("localhost", port);
+    server.start();
+    final HTTPRequestBuilder hrb = new HTTPRequestBuilder(new URL("http://localhost:"+port));
+    hrb.setBody(IOUtils.EMPTY_BYTEBUFFER);
+    final HTTPClient httpClient = new HTTPClient(1, 1048576, 1) {
+      @Override
+      protected void processQueue() {
+        // queue is never processed so we know it's queued
+      }
+    };
+    httpClient.start();
+    httpClient.requestAsync(hrb.buildClientHTTPRequest()); // first request should queue fine
+    try{
+      httpClient.request(hrb.buildClientHTTPRequest());
+      fail();
+    } catch(RejectedExecutionException e) {
+      assertEquals(1, httpClient.getRequestQueueSize());
+      assertEquals("Request queue full", e.getMessage());
+    }
+  }
 
   @Test
   public void expireRequest() throws IOException, HTTPParsingException {
@@ -356,8 +382,8 @@ public void expireRequest() throws IOException, HTTPParsingException {
     try {
       httpClient.request(hrb.buildClientHTTPRequest());
       fail();
-    } catch(HTTPParsingException hp) {
-
+    } catch(CancellationException e) {
+      assertEquals("Request timed out at point: ReadingResponseBody", e.getMessage());
     }
     assertTrue((Clock.accurateForwardProgressingMillis() - start) > 30);
   }

gradle.properties

@@ -1,5 +1,5 @@
 group = org.threadly
-version = 0.23
+version = 0.24-SNAPSHOT
 threadlyVersion = 5.37
 litesocketsVersion = 4.10
 org.gradle.parallel=false