HTTPClient: Change how response bodies are handled

This is an API breaking change to change how response bodies are handled.
Instead of having the HTTPClient handle the body, the body handling is instead put into the request (so each request may handle the body differently if needed).
This breaks the API in that HTTPClient is no longer monitoring the maximum body accepted, but rather the default behavior in HTTPRequestBuilder is to buffer and monitor the body limit that was previously done in HTTPClient.

Author: jentfoo

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

@@ -60,12 +60,10 @@
  * is kept in memory and are not handled as streams.  See {@link HTTPStreamClient} for use with large HTTP data sets.</p>
  */
 public class HTTPClient extends AbstractService {
-  public static final int DEFAULT_CONCURRENT = 2;
+  public static final int DEFAULT_CONCURRENT = 8;
   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 Queue<HTTPRequestWrapper> queue;
@@ -90,18 +88,17 @@ public class HTTPClient extends AbstractService {
    * 
    */
   public HTTPClient() {
-    this(DEFAULT_CONCURRENT, MAX_HTTP_RESPONSE);
+    this(DEFAULT_CONCURRENT, -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>
-   *  
+   * 
    * @param maxConcurrent maximum number of requests to run simultaneously. 
-   * @param maxResponseSize the maximum responseSize clients are allowed to send.
    */
-  public HTTPClient(int maxConcurrent, int maxResponseSize) {
-    this(maxConcurrent, maxResponseSize, -1);
+  public HTTPClient(int maxConcurrent) {
+    this(maxConcurrent, -1);
   }
 
   /**
@@ -114,12 +111,10 @@ public HTTPClient(int maxConcurrent, int maxResponseSize) {
    * 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) {
+  public HTTPClient(int maxConcurrent, int maxQueueSize) {
     this.maxConcurrent = maxConcurrent;
-    this.maxResponseSize = maxResponseSize;
     sts = new SingleThreadScheduler();
     this.ssi = sts;
     ntse = new NoThreadSocketExecuter();
@@ -137,11 +132,10 @@ public HTTPClient(int maxConcurrent, int maxResponseSize, int maxQueueSize) {
    * as well as your own {@link SocketExecuter} as the thread pool to use.</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.
    */
-  public HTTPClient(int maxConcurrent, int maxResponseSize, SocketExecuter sei) {
-    this(maxConcurrent, maxResponseSize, sei, -1);
+  public HTTPClient(int maxConcurrent, SocketExecuter sei) {
+    this(maxConcurrent, -1, sei);
   }
 
   /**
@@ -153,14 +147,12 @@ public HTTPClient(int maxConcurrent, int maxResponseSize, SocketExecuter sei) {
    * 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 maxConcurrent maximum number of requests to run simultaneously.
    * @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) {
+  public HTTPClient(int maxConcurrent, int maxQueueSize, SocketExecuter sei) {
     this.maxConcurrent = maxConcurrent;
-    this.maxResponseSize = maxResponseSize;
     this.ssi = sei.getThreadScheduler();
     this.sei = sei;
     if (maxQueueSize < 1 || maxQueueSize == Integer.MAX_VALUE) {
@@ -187,7 +179,7 @@ public int getRequestQueueSize() {
    * 
    * @return number of request currently in progress.
    */
-  public int getInProgressSize() {
+  public int getInProgressCount() {
     return this.inProcess.size();
   }
 
@@ -196,7 +188,7 @@ public int getInProgressSize() {
    * 
    * @return number of open connections.
    */
-  public int getOpenConnections() {
+  public int getOpenConnectionCount() {
     return tcpClients.size();
   }
 
@@ -634,7 +626,6 @@ private class HTTPRequestWrapper implements HTTPResponseCallback {
     private final ClientHTTPRequest chr;
     private RequestState currentState = RequestState.Queued;
     private HTTPResponse response;
-    private ReuseableMergedByteBuffers responseMBB = new ReuseableMergedByteBuffers();
     private TCPClient client;
     private long lastRead = Clock.lastKnownForwardProgressingMillis();
 
@@ -685,10 +676,11 @@ public void headersFinished(HTTPResponse hr) {
 
     @Override
     public void bodyData(ByteBuffer bb) {
-      responseMBB.add(bb);
-      if(responseMBB.remaining() > maxResponseSize) {
+      try {
+        chr.getBodyConsumer().accept(bb);
+      } catch (Exception e) {
         TCPClient client = this.client; // will be `null` after error
-        slf.setFailure(new HTTPParsingException("Response Body to large!"));
+        slf.setFailure(e);
         client.close();
       }
     }
@@ -697,7 +689,7 @@ public void bodyData(ByteBuffer bb) {
     public void finished() {
       currentState = RequestState.Finished;
       slf.setResult(new HTTPResponseData(HTTPClient.this, chr.getHTTPRequest(), response, 
-                                         responseMBB.duplicateAndClean()));
+                                         chr.getBodyConsumer().finishBody()));
       hrp.removeHTTPResponseCallback(this);
       TCPClient client = this.client;
       this.client = null;
@@ -766,7 +758,7 @@ public HTTPResponseData(HTTPClient client, HTTPRequest origRequest, HTTPResponse
                             MergedByteBuffers bb) {
       this.client = client;
       this.hr = hr;
-      this.body = bb;
+      this.body = bb == null ? new SimpleMergedByteBuffers(false) : bb;
       this.origRequest = origRequest;
     }
 

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

@@ -15,6 +15,7 @@
 import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import org.junit.After;
 import org.junit.Before;
@@ -29,8 +30,10 @@
 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.client.http.HTTPClient.HTTPResponseData;
 import org.threadly.litesockets.protocols.http.request.ClientHTTPRequest;
+import org.threadly.litesockets.protocols.http.request.ClientHTTPRequest.BodyConsumer;
 import org.threadly.litesockets.protocols.http.request.HTTPRequestBuilder;
 import org.threadly.litesockets.protocols.http.response.HTTPResponse;
 import org.threadly.litesockets.protocols.http.response.HTTPResponseBuilder;
@@ -58,7 +61,7 @@ public class HTTPClientTests {
   static HTTPResponse RESPONSE_HUGE_NOCL;
   static {
     StringBuilder sb = new StringBuilder();
-    while(sb.length() < (HTTPClient.MAX_HTTP_RESPONSE*2)) {
+    while(sb.length() < (HTTPRequestBuilder.MAX_HTTP_BUFFERED_RESPONSE*2)) {
       sb.append(CONTENT);
     }
     LARGE_CONTENT = sb.toString();
@@ -146,7 +149,7 @@ public void manyRequestsConcurrentJavaExecutor() throws IOException, Interrupted
     fakeServer = new TestHTTPServer(port, RESPONSE_CL, CONTENT, false, false);
     final HTTPRequestBuilder hrb = new HTTPRequestBuilder(new URL("http://localhost:"+port));
     hrb.setHTTPAddress(new HTTPAddress("localhost", port, false), true);
-    final HTTPClient httpClient = new HTTPClient(HTTPClient.DEFAULT_CONCURRENT, HTTPClient.MAX_HTTP_RESPONSE, TSE);
+    final HTTPClient httpClient = new HTTPClient(HTTPClient.DEFAULT_CONCURRENT, TSE);
     httpClient.start();
 
     AsyncVerifier av = new AsyncVerifier();
@@ -192,7 +195,7 @@ public void manyRequestsConcurrentOnPool() throws IOException, InterruptedExcept
     hrb.setHTTPAddress(new HTTPAddress("localhost", port, false), true);
     final ThreadedSocketExecuter TSE = new ThreadedSocketExecuter(PS);
     TSE.start();
-    final HTTPClient httpClient = new HTTPClient(200, HTTPClient.MAX_HTTP_RESPONSE, TSE);
+    final HTTPClient httpClient = new HTTPClient(200, TSE);
     httpClient.start();
 
     AsyncVerifier av = new AsyncVerifier();
@@ -292,6 +295,32 @@ public void streamedBodyRequest() throws IOException, HTTPParsingException, Inte
     assertEquals(CONTENT, lf.get().getBodyAsString());
   }
 
+  @Test
+  public void streamedBodyResponse() throws IOException, HTTPParsingException {
+    int port = PortUtils.findTCPPort();
+    fakeServer = new TestHTTPServer(port, RESPONSE_HUGE, LARGE_CONTENT, false, false);
+    AtomicInteger readContentSize = new AtomicInteger(0);
+    final HTTPRequestBuilder hrb = new HTTPRequestBuilder(new URL("http://localhost:"+port))
+        .setBodyConsumer(new BodyConsumer() {
+          @Override
+          public void accept(ByteBuffer bb) throws HTTPParsingException {
+            readContentSize.addAndGet(bb.remaining());
+          }
+
+          @Override
+          public MergedByteBuffers finishBody() {
+            return null;
+          }
+        });
+    
+    final HTTPClient httpClient = new HTTPClient();
+    httpClient.start();
+    MergedByteBuffers emptyResponseBody = httpClient.request(hrb.buildClientHTTPRequest()).getBody();
+    
+    assertEquals(0, emptyResponseBody.remaining());
+    assertEquals(LARGE_CONTENT.length(), readContentSize.get());
+  }
+
   @Test
   public void contentLengthOnHeadRequest() throws IOException, HTTPParsingException {
     int port = PortUtils.findTCPPort();
@@ -342,7 +371,7 @@ public void timeoutRequest() throws IOException, HTTPParsingException {
       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);
+      new TestCondition(() -> httpClient.getInProgressCount() == 0).blockTillTrue(1_000);
     }
   }
 
@@ -368,7 +397,7 @@ protected void processQueue() {
       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);
+      new TestCondition(() -> httpClient.getInProgressCount() == 0).blockTillTrue(1_000);
     }
   }
 
@@ -379,7 +408,7 @@ public void queueLimitTest() throws IOException, HTTPParsingException {
     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) {
+    final HTTPClient httpClient = new HTTPClient(1, 1) {
       @Override
       protected void processQueue() {
         // queue is never processed so we know it's queued
@@ -426,7 +455,7 @@ public void sslRequest() throws IOException, HTTPParsingException {
   }
 
   @Test
-  public void toLargeRequest() throws IOException, HTTPParsingException {
+  public void tooLargeRequest() throws IOException, HTTPParsingException {
     int port = PortUtils.findTCPPort();
     fakeServer = new TestHTTPServer(port, RESPONSE_HUGE, LARGE_CONTENT, false, false);
     final HTTPRequestBuilder hrb = new HTTPRequestBuilder(new URL("http://localhost:"+port));
@@ -441,7 +470,7 @@ public void toLargeRequest() throws IOException, HTTPParsingException {
   }
 
   @Test
-  public void toLargeRequestNoContentLength() throws IOException, HTTPParsingException {
+  public void tooLargeRequestNoContentLength() throws IOException, HTTPParsingException {
     int port = PortUtils.findTCPPort();
     fakeServer = new TestHTTPServer(port, RESPONSE_NO_CL, LARGE_CONTENT, false, true);
     final HTTPRequestBuilder hrb = new HTTPRequestBuilder(new URL("http://localhost:"+port));

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

@@ -5,7 +5,9 @@
 
 import org.threadly.concurrent.future.FutureUtils;
 import org.threadly.concurrent.future.ListenableFuture;
+import org.threadly.litesockets.buffers.MergedByteBuffers;
 import org.threadly.litesockets.protocols.http.shared.HTTPAddress;
+import org.threadly.litesockets.protocols.http.shared.HTTPParsingException;
 
 // TODO - do we want to move this into the `client`?  I see the `HTTPRequestBuilder` references it, 
 // but we could create an extending class `ClientHTTPRequestBuilder` which can build this
@@ -20,13 +22,16 @@ public class ClientHTTPRequest {
   private final HTTPRequest request;
   private final HTTPAddress ha;
   private final Supplier<ListenableFuture<ByteBuffer>> bodyProvider;
+  private final BodyConsumer bodyConsumer;
   private final int timeoutMS; 
 
   protected ClientHTTPRequest(HTTPRequest request, HTTPAddress ha, int timeoutMS, 
-                              Supplier<ListenableFuture<ByteBuffer>> bodyProvider) {
+                              Supplier<ListenableFuture<ByteBuffer>> bodyProvider, 
+                              BodyConsumer bodyConsumer) {
     this.request = request;
     this.ha = ha;
     this.bodyProvider = bodyProvider == null ? EMPTY_BODY_SUPPLIER : bodyProvider;
+    this.bodyConsumer = bodyConsumer;
     this.timeoutMS = timeoutMS;
   }
 
@@ -43,6 +48,10 @@ public HTTPAddress getHTTPAddress() {
     return ha;
   }
 
+  public BodyConsumer getBodyConsumer() {
+    return bodyConsumer;
+  }
+  
   /**
    * Returns if there is a body associated to this request.
    * 
@@ -64,14 +73,6 @@ public ListenableFuture<ByteBuffer> nextBodySection() {
   public int getTimeoutMS() {
     return this.timeoutMS;
   }
-
-  // TODO - is this useful?  I am not seeing any cases of it being used
-  public HTTPRequestBuilder makeBuilder() {
-    HTTPRequestBuilder hrb = request.makeBuilder();
-    hrb.setHTTPAddress(ha, false);
-        
-    return hrb;
-  }
 
   @Override
   public int hashCode() {
@@ -112,4 +113,26 @@ public boolean equals(Object obj) {
     }
     return true;
   }
+  
+  /**
+   * Consumer to accept the response body as it is read from the network.
+   */
+  public interface BodyConsumer {
+    /**
+     * Invoked to provide newly read body contents.
+     * 
+     * @param bb Buffer containing the body data
+     * @throws HTTPParsingException May be thrown if there is any errors in the body contents
+     */
+    public void accept(ByteBuffer bb) throws HTTPParsingException;
+    
+    /**
+     * Invoked once the body has been fully consumed.  This should finish and cleanup anything 
+     * necessary, providing a {@link MergedByteBuffers} to be represented in the final 
+     * {@link HTTPResponseData}.
+     * 
+     * @return Buffer containing the final form of the body (or empty if body wont be read from {@link HTTPResponseData})
+     */
+    public MergedByteBuffers finishBody();
+  }
 }