HTTPClient: Slight tweaks

Makes sure that errors loading body data are communicated to the response future.
This also will start loading the next data load while the write is still sending, but will block to load further till the write has completed.

Author: jentfoo

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

@@ -23,6 +23,7 @@
 import org.threadly.concurrent.ReschedulingOperation;
 import org.threadly.concurrent.SingleThreadScheduler;
 import org.threadly.concurrent.SubmitterScheduler;
+import org.threadly.concurrent.future.FutureCallback;
 import org.threadly.concurrent.future.FutureUtils;
 import org.threadly.concurrent.future.ListenableFuture;
 import org.threadly.concurrent.future.SettableListenableFuture;
@@ -401,23 +402,28 @@ private void process(HTTPRequestWrapper hrw) {
         if (hrw.chr.getHTTPRequest().getHTTPHeaders().isChunked()) {
           hrw.client.write(hrw.chr.getHTTPRequest().getMergedByteBuffers());
 
-          hrw.chr.nextBodySection().resultCallback(new Consumer<ByteBuffer>() {
+          hrw.chr.nextBodySection().callback(new FutureCallback<ByteBuffer>() {
             @Override
-            public void accept(ByteBuffer bb) {
+            public void handleResult(ByteBuffer bb) {
               ListenableFuture<?> writeFuture = hrw.client.write(HTTPUtils.wrapInChunk(bb));
 
               if (bb != null && bb.hasRemaining()) {
-                writeFuture.resultCallback((ignored) -> 
-                    hrw.chr.nextBodySection().resultCallback(this));
+                ListenableFuture<ByteBuffer> nextWrite = hrw.chr.nextBodySection();
+                writeFuture.resultCallback((ignored) -> nextWrite.callback(this));
               }
             }
+
+            @Override
+            public void handleFailure(Throwable t) {
+              hrw.slf.handleFailure(t);
+            }
           });
         } else {
-          hrw.chr.nextBodySection().resultCallback(new Consumer<ByteBuffer>() {
+          hrw.chr.nextBodySection().callback(new FutureCallback<ByteBuffer>() {
             private boolean firstSection = true;
 
             @Override
-            public void accept(ByteBuffer bb) {
+            public void handleResult(ByteBuffer bb) {
               if (bb != null && bb.hasRemaining()) {
                 MergedByteBuffers writeBuffer;
                 if (firstSection) {
@@ -429,13 +435,19 @@ public void accept(ByteBuffer bb) {
                   writeBuffer = new SimpleMergedByteBuffers(false, bb);
                 }
 
+                ListenableFuture<ByteBuffer> nextWrite = hrw.chr.nextBodySection();
                 hrw.client.write(writeBuffer)
-                          .resultCallback((ignored) -> hrw.chr.nextBodySection().resultCallback(this));
+                          .resultCallback((ignored) -> nextWrite.callback(this));
               } else if (firstSection) {
                 firstSection = false;
                 hrw.client.write(hrw.chr.getHTTPRequest().getMergedByteBuffers());
               }
             }
+
+            @Override
+            public void handleFailure(Throwable t) {
+              hrw.slf.handleFailure(t);
+            }
           });
         }
       } else {