Some cleanup / unit test fixes

I forgot to re-enable the new unit test which required some uncommitted TestHTTPServer changes.
The test is dependent on the TestHTTPServer not sending the response till the request has completed.

Author: jentfoo

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

@@ -15,7 +15,6 @@
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.TimeUnit;
-import java.util.function.Consumer;
 
 import javax.net.ssl.SSLContext;
 import javax.net.ssl.SSLEngine;
@@ -83,7 +82,7 @@ public class HTTPClient extends AbstractService {
 
   /**
    * <p>This is the default constructor it will create its own {@link SingleThreadScheduler} to use as a threadpool, as
-   * well as using the default {@value DEFAULT_CONCURRENT} and {@value MAX_HTTP_RESPONSE}.  This means we will do at most
+   * well as using the default {@value DEFAULT_CONCURRENT} and no queue limit.  This means we will do at most
    * 2 HTTPRequests at the same time, and those responses can be up to 1mb in size.</p>
    * 
    */

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

@@ -45,10 +45,8 @@
 import org.threadly.litesockets.utils.PortUtils;
 import org.threadly.test.concurrent.AsyncVerifier;
 import org.threadly.test.concurrent.TestCondition;
-import org.threadly.test.concurrent.TestUtils;
 import org.threadly.util.ArrayIterator;
 import org.threadly.util.Clock;
-import org.threadly.util.debug.Profiler;
 
 public class HTTPClientTests {
   static String CONTENT = "TEST123";
@@ -107,13 +105,13 @@ public void manyRequestsConcurrent() throws IOException, InterruptedException, T
       @Override
       public void run() {
         ClientHTTPRequest chr = hrb.buildClientHTTPRequest();
-        final long start = Clock.accurateForwardProgressingMillis();
+        //final long start = Clock.accurateForwardProgressingMillis();
 
         final ListenableFuture<HTTPResponseData>  lf = httpClient.requestAsync(chr);
         lf.callback(new FutureCallback<HTTPResponseData>() {
           @Override
           public void handleResult(HTTPResponseData result) {
-            System.out.println("DELAY:"+(Clock.accurateForwardProgressingMillis()-start));
+            //System.out.println("DELAY:"+(Clock.accurateForwardProgressingMillis()-start));
             av.assertEquals(CONTENT, result.getBodyAsString());
             av.signalComplete();
           }
@@ -127,13 +125,6 @@ public void handleFailure(Throwable t) {
       for(int i=0; i<number; i++) {
         CLIENT_PS.execute(run);
       }
-
-      Profiler p = new Profiler();
-      TestUtils.sleep(2_000);
-      p.start();
-      TestUtils.sleep(2_000);
-      p.stop();
-      System.out.println(p.dump());
 
       av.waitForTest(10_000, number);
 
@@ -234,7 +225,7 @@ public void handleFailure(Throwable t) {
   @Test
   public void blockingRequest() throws IOException, HTTPParsingException {
     int port = PortUtils.findTCPPort();
-    fakeServer = new TestHTTPServer(port, RESPONSE_CL, CONTENT, false, true);
+    fakeServer = new TestHTTPServer(port, RESPONSE_CL, CONTENT, false, true, true);
     final HTTPRequestBuilder hrb = new HTTPRequestBuilder(new URL("http://localhost:"+port));
     hrb.setHTTPAddress(new HTTPAddress("localhost", port, false), true);
     final HTTPClient httpClient = new HTTPClient();
@@ -257,7 +248,7 @@ public void noContentLengthNoBody() throws IOException, HTTPParsingException {
   @Test
   public void noContentLengthWithBody() throws IOException, HTTPParsingException {
     int port = PortUtils.findTCPPort();
-    fakeServer = new TestHTTPServer(port, RESPONSE_NO_CL, CONTENT, false, true);
+    fakeServer = new TestHTTPServer(port, RESPONSE_NO_CL, CONTENT, false, true, true);
     final HTTPRequestBuilder hrb = new HTTPRequestBuilder(new URL("http://localhost:"+port));
 
     final HTTPClient httpClient = new HTTPClient();
@@ -267,10 +258,10 @@ public void noContentLengthWithBody() throws IOException, HTTPParsingException {
     assertEquals(CONTENT, hrs.getBodyAsString());
   }
 
-  //@Test
+  @Test
   public void streamedBodyRequest() throws IOException, HTTPParsingException, InterruptedException, ExecutionException {
     int port = PortUtils.findTCPPort();
-    fakeServer = new TestHTTPServer(port, RESPONSE_CL, CONTENT, false, true);
+    fakeServer = new TestHTTPServer(port, RESPONSE_CL, CONTENT, false, true, true);
     ByteBuffer write1 = ByteBuffer.allocate(100);
     ByteBuffer write2 = ByteBuffer.allocate(100);
     SettableListenableFuture<ByteBuffer> write1SLF = new SettableListenableFuture<>();
@@ -298,7 +289,7 @@ public void streamedBodyRequest() throws IOException, HTTPParsingException, Inte
   @Test
   public void streamedBodyResponse() throws IOException, HTTPParsingException {
     int port = PortUtils.findTCPPort();
-    fakeServer = new TestHTTPServer(port, RESPONSE_HUGE, LARGE_CONTENT, false, false);
+    fakeServer = new TestHTTPServer(port, RESPONSE_HUGE, LARGE_CONTENT, false, false, false);
     AtomicInteger readContentSize = new AtomicInteger(0);
     final HTTPRequestBuilder hrb = new HTTPRequestBuilder(new URL("http://localhost:"+port))
         .setBodyConsumer(new BodyConsumer() {
@@ -349,8 +340,7 @@ public void closeBeforeLength() throws IOException, HTTPParsingException {
       httpClient.request(hrb.buildClientHTTPRequest());
       fail();
     } catch(HTTPParsingException e) {
-      e.printStackTrace();
-      assertEquals("Did not get complete body!", e.getMessage());
+      assertTrue(e.getMessage().startsWith("Body not complete"));
     }
   }
 
@@ -546,7 +536,7 @@ public void accept(Client c) {
   @Test
   public void urlRequest() throws HTTPParsingException, IOException {
     int port = PortUtils.findTCPPort();
-    fakeServer = new TestHTTPServer(port, RESPONSE_CL, CONTENT, false, false);
+    fakeServer = new TestHTTPServer(port, RESPONSE_CL, CONTENT, false, false, true);
     final HTTPClient httpClient = new HTTPClient();
     httpClient.start();
     HTTPResponseData hrd = httpClient.request(new URL("http://localhost:"+port));

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

@@ -22,6 +22,7 @@
 import org.threadly.litesockets.protocols.http.response.HTTPResponse;
 import org.threadly.litesockets.protocols.websocket.WSFrame;
 import org.threadly.litesockets.utils.SSLUtils.FullTrustManager;
+import org.threadly.util.ExceptionUtils;
 
 public class TestHTTPServer {
   private final TrustManager[] myTMs = new TrustManager [] {new FullTrustManager() };
@@ -35,9 +36,15 @@ public class TestHTTPServer {
   private TCPServer server;
   private TransactionalByteBuffers sendBack = new TransactionalByteBuffers();
   private boolean closeOnSend;
+  private boolean delayResponse;
   private HTTPResponse hr;
 
   public TestHTTPServer(int port, HTTPResponse hr, String sendBack, boolean doSSL, boolean closeOnSend) throws IOException {
+    this(port, hr, sendBack, doSSL, closeOnSend, false);
+  }
+
+  public TestHTTPServer(int port, HTTPResponse hr, String sendBack, boolean doSSL, 
+                        boolean closeOnSend, boolean delayResponse) throws IOException {
     if(doSSL) {
       doSSLCrap();
     } 
@@ -53,6 +60,7 @@ public TestHTTPServer(int port, HTTPResponse hr, String sendBack, boolean doSSL,
     server.start();
     this.sendBack.add(ByteBuffer.wrap(sendBack.getBytes()));
     this.closeOnSend = closeOnSend;
+    this.delayResponse = delayResponse;
   }
 
   private void doSSLCrap() {
@@ -81,9 +89,7 @@ public void stop() {
   private void onClient(final Client c) {
     final HTTPRequestProcessor hrp = new HTTPRequestProcessor();
     hrp.addHTTPRequestCallback(new HTTPRequestCallback() {
-
-      @Override
-      public void headersFinished(HTTPRequest hreq) {
+      private void sendResponse() {
         ListenableFuture<?> lf = c.write(hr.getMergedByteBuffers());
         sendBack.begin();
 
@@ -94,31 +100,35 @@ public void headersFinished(HTTPRequest hreq) {
         if(closeOnSend) {
           lf.listener(c::close);
         }
-        hrp.reset();
+      }
+
+      @Override
+      public void headersFinished(HTTPRequest hreq) {
+        if (! delayResponse) {
+          sendResponse();
+        }
       }
 
       @Override
       public void bodyData(ByteBuffer bb) {
-        // TODO Auto-generated method stub
-        
+        // ignored
       }
 
       @Override
       public void websocketData(WSFrame wsf, ByteBuffer bb) {
-        // TODO Auto-generated method stub
-        
+        // ignored
       }
 
       @Override
       public void finished() {
-        // TODO Auto-generated method stub
-        
+        if (delayResponse) {
+          sendResponse();
+        }
       }
 
       @Override
       public void hasError(Throwable t) {
-        // TODO Auto-generated method stub
-        
+        ExceptionUtils.handleException(t);
       }});
     clients.putIfAbsent(c, hrp);
     c.setReader((lc)->hrp.processData(lc.getRead()));

protocol/src/main/java/org/threadly/litesockets/protocols/http/response/HTTPResponseProcessor.java

@@ -138,8 +138,9 @@ public void connectionClosed() {
           reset(new HTTPParsingException("Did not complete chunked encoding!"));
         }
       } else {
-        if(response.getHeaders().getContentLength() > 0 && response.getHeaders().getContentLength() != this.currentBodySize) {
-          reset(new HTTPParsingException("Did not get complete body!"));
+        long contentLength = response.getHeaders().getContentLength();
+        if(contentLength > 0 && contentLength != this.currentBodySize) {
+          reset(new HTTPParsingException("Body not completed! (" + currentBodySize + "/" + contentLength + ")"));
         } else {
           reset(null);
         }