Make TransferListener. onBytesReceived receive a byte array

Author: slandelle

api/src/main/java/org/asynchttpclient/listener/TransferCompletionHandler.java

@@ -21,7 +21,6 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
-import java.nio.ByteBuffer;
 import java.util.concurrent.ConcurrentLinkedQueue;
 
 /**
@@ -122,7 +121,7 @@ public STATE onBodyPartReceived(final HttpResponseBodyPart content) throws Excep
         if (accumulateResponseBytes) {
             s = super.onBodyPartReceived(content);
         }
-        fireOnBytesReceived(content.getBodyByteBuffer());
+        fireOnBytesReceived(content.getBodyPartBytes());
         return s;
     }
 
@@ -187,7 +186,7 @@ private void fireOnEnd() {
         }
     }
 
-    private void fireOnBytesReceived(ByteBuffer b) {
+    private void fireOnBytesReceived(byte[] b) {
         for (TransferListener l : listeners) {
             try {
                 l.onBytesReceived(b);

api/src/main/java/org/asynchttpclient/listener/TransferListener.java

@@ -15,7 +15,6 @@
 import org.asynchttpclient.FluentCaseInsensitiveStringsMap;
 
 import java.io.IOException;
-import java.nio.ByteBuffer;
 
 /**
  * A simple interface an application can implements in order to received byte transfer information.
@@ -35,9 +34,9 @@ public interface TransferListener {
     /**
      * Invoked every time response's chunk are received.
      *
-     * @param buffer a {@link ByteBuffer}
+     * @param bytes a {@link byte[]}
      */
-    public void onBytesReceived(ByteBuffer buffer) throws IOException;
+    public void onBytesReceived(byte[] bytes) throws IOException;
 
     /**
      * Invoked every time request's chunk are sent.

api/src/test/java/org/asynchttpclient/async/TransferListenerTest.java

@@ -12,35 +12,46 @@
  */
 package org.asynchttpclient.async;
 
-import org.asynchttpclient.AsyncHttpClient;
-import org.asynchttpclient.FluentCaseInsensitiveStringsMap;
-import org.asynchttpclient.Response;
-import org.asynchttpclient.generators.FileBodyGenerator;
-import org.asynchttpclient.listener.TransferCompletionHandler;
-import org.asynchttpclient.listener.TransferListener;
-import org.eclipse.jetty.server.handler.AbstractHandler;
-import org.testng.annotations.Test;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.fail;
 
-import javax.servlet.ServletException;
-import javax.servlet.http.HttpServletRequest;
-import javax.servlet.http.HttpServletResponse;
 import java.io.File;
 import java.io.FileOutputStream;
 import java.io.IOException;
-import java.nio.ByteBuffer;
+import java.nio.charset.Charset;
 import java.util.Enumeration;
 import java.util.UUID;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
 
-import static org.testng.Assert.assertEquals;
-import static org.testng.Assert.assertNotNull;
-import static org.testng.Assert.assertNull;
-import static org.testng.Assert.fail;
+import javax.servlet.ServletException;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+
+import org.asynchttpclient.AsyncHttpClient;
+import org.asynchttpclient.AsyncHttpClientConfig;
+import org.asynchttpclient.FluentCaseInsensitiveStringsMap;
+import org.asynchttpclient.Response;
+import org.asynchttpclient.generators.FileBodyGenerator;
+import org.asynchttpclient.listener.TransferCompletionHandler;
+import org.asynchttpclient.listener.TransferListener;
+import org.eclipse.jetty.server.handler.AbstractHandler;
+import org.testng.Assert;
+import org.testng.annotations.Test;
 
 public abstract class TransferListenerTest extends AbstractBasicTest {
+
     private static final File TMP = new File(System.getProperty("java.io.tmpdir"), "ahc-tests-" + UUID.randomUUID().toString().substring(0, 8));
+    private static final byte[] PATTERN_BYTES = "RatherLargeFileRatherLargeFileRatherLargeFileRatherLargeFile".getBytes(Charset.forName("UTF-16"));
+
+    static {
+        TMP.mkdirs();
+        TMP.deleteOnExit();
+    }
 
     private class BasicHandler extends AbstractHandler {
 
@@ -81,7 +92,7 @@ public void basicGetTest() throws Throwable {
             final AtomicReference<Throwable> throwable = new AtomicReference<Throwable>();
             final AtomicReference<FluentCaseInsensitiveStringsMap> hSent = new AtomicReference<FluentCaseInsensitiveStringsMap>();
             final AtomicReference<FluentCaseInsensitiveStringsMap> hRead = new AtomicReference<FluentCaseInsensitiveStringsMap>();
-            final AtomicReference<ByteBuffer> bb = new AtomicReference<ByteBuffer>();
+            final AtomicReference<byte[]> bb = new AtomicReference<byte[]>();
             final AtomicBoolean completed = new AtomicBoolean(false);
 
             TransferCompletionHandler tl = new TransferCompletionHandler();
@@ -95,11 +106,11 @@ public void onResponseHeadersReceived(FluentCaseInsensitiveStringsMap headers) {
                     hRead.set(headers);
                 }
 
-                public void onBytesReceived(ByteBuffer buffer) {
-                    bb.set(buffer);
+                public void onBytesReceived(byte[] b) {
+                    bb.set(b);
                 }
 
-                public void onBytesSent(ByteBuffer buffer) {
+                public void onBytesSent(long amount, long current, long total) {
                 }
 
                 public void onRequestResponseCompleted() {
@@ -130,20 +141,24 @@ public void onThrowable(Throwable t) {
 
     @Test(groups = { "standalone", "default_provider" })
     public void basicPutTest() throws Throwable {
-        AsyncHttpClient c = getAsyncHttpClient(null);
-        try {
-            final AtomicReference<Throwable> throwable = new AtomicReference<Throwable>();
-            final AtomicReference<FluentCaseInsensitiveStringsMap> hSent = new AtomicReference<FluentCaseInsensitiveStringsMap>();
-            final AtomicReference<FluentCaseInsensitiveStringsMap> hRead = new AtomicReference<FluentCaseInsensitiveStringsMap>();
-            final AtomicInteger bbReceivedLenght = new AtomicInteger(0);
-            final AtomicInteger bbSentLenght = new AtomicInteger(0);
+        final AtomicReference<Throwable> throwable = new AtomicReference<Throwable>();
+        final AtomicReference<FluentCaseInsensitiveStringsMap> hSent = new AtomicReference<FluentCaseInsensitiveStringsMap>();
+        final AtomicReference<FluentCaseInsensitiveStringsMap> hRead = new AtomicReference<FluentCaseInsensitiveStringsMap>();
+        final AtomicInteger bbReceivedLenght = new AtomicInteger(0);
+        final AtomicLong bbSentLenght = new AtomicLong(0L);
 
-            final AtomicBoolean completed = new AtomicBoolean(false);
+        final AtomicBoolean completed = new AtomicBoolean(false);
+
+        byte[] bytes = "RatherLargeFileRatherLargeFileRatherLargeFileRatherLargeFile".getBytes("UTF-16");
+        long repeats = (1024 * 100 * 10 / bytes.length) + 1;
+        File file = createTempFile(bytes, (int) repeats);
+        long expectedFileSize = PATTERN_BYTES.length * repeats;
+        Assert.assertEquals(expectedFileSize, file.length(), "Invalid file length");
 
-            byte[] bytes = "RatherLargeFileRatherLargeFileRatherLargeFileRatherLargeFile".getBytes("UTF-16");
-            long repeats = (1024 * 100 * 10 / bytes.length) + 1;
-            File largeFile = createTempFile(bytes, (int) repeats);
+        int timeout = (int) (repeats / 1000);
+        AsyncHttpClient client = getAsyncHttpClient(new AsyncHttpClientConfig.Builder().setConnectionTimeoutInMs(timeout).build());
 
+        try {
             TransferCompletionHandler tl = new TransferCompletionHandler();
             tl.addTransferListener(new TransferListener() {
 
@@ -155,12 +170,12 @@ public void onResponseHeadersReceived(FluentCaseInsensitiveStringsMap headers) {
                     hRead.set(headers);
                 }
 
-                public void onBytesReceived(ByteBuffer buffer) {
-                    bbReceivedLenght.addAndGet(buffer.capacity());
+                public void onBytesReceived(byte[] b) {
+                    bbReceivedLenght.addAndGet(b.length);
                 }
 
-                public void onBytesSent(ByteBuffer buffer) {
-                    bbSentLenght.addAndGet(buffer.capacity());
+                public void onBytesSent(long amount, long current, long total) {
+                    bbSentLenght.addAndGet(amount);
                 }
 
                 public void onRequestResponseCompleted() {
@@ -173,37 +188,38 @@ public void onThrowable(Throwable t) {
             });
 
             try {
-                Response response = c.preparePut(getTargetUrl()).setBody(largeFile).execute(tl).get();
+                Response response = client.preparePut(getTargetUrl()).setBody(file).execute(tl).get();
 
                 assertNotNull(response);
                 assertEquals(response.getStatusCode(), 200);
                 assertNotNull(hRead.get());
                 assertNotNull(hSent.get());
-                assertEquals(bbReceivedLenght.get(), largeFile.length());
-                assertEquals(bbSentLenght.get(), largeFile.length());
+                assertEquals(bbReceivedLenght.get(), expectedFileSize, "Number of received bytes incorrect");
+                assertEquals(bbSentLenght.get(), expectedFileSize, "Number of sent bytes incorrect");
             } catch (IOException ex) {
                 fail("Should have timed out");
             }
         } finally {
-            c.close();
+            client.close();
         }
     }
 
     @Test(groups = { "standalone", "default_provider" })
     public void basicPutBodyTest() throws Throwable {
-        AsyncHttpClient c = getAsyncHttpClient(null);
+        AsyncHttpClient client = getAsyncHttpClient(null);
         try {
             final AtomicReference<Throwable> throwable = new AtomicReference<Throwable>();
             final AtomicReference<FluentCaseInsensitiveStringsMap> hSent = new AtomicReference<FluentCaseInsensitiveStringsMap>();
             final AtomicReference<FluentCaseInsensitiveStringsMap> hRead = new AtomicReference<FluentCaseInsensitiveStringsMap>();
             final AtomicInteger bbReceivedLenght = new AtomicInteger(0);
-            final AtomicInteger bbSentLenght = new AtomicInteger(0);
+            final AtomicLong bbSentLenght = new AtomicLong(0L);
 
             final AtomicBoolean completed = new AtomicBoolean(false);
 
-            byte[] bytes = "RatherLargeFileRatherLargeFileRatherLargeFileRatherLargeFile".getBytes("UTF-16");
-            long repeats = (1024 * 100 * 10 / bytes.length) + 1;
-            File largeFile = createTempFile(bytes, (int) repeats);
+            long repeats = (1024 * 100 * 10 / PATTERN_BYTES.length) + 1;
+            File file = createTempFile(PATTERN_BYTES, (int) repeats);
+            long expectedFileSize = PATTERN_BYTES.length * repeats;
+            Assert.assertEquals(expectedFileSize, file.length(), "Invalid file length");
 
             TransferCompletionHandler tl = new TransferCompletionHandler();
             tl.addTransferListener(new TransferListener() {
@@ -216,12 +232,12 @@ public void onResponseHeadersReceived(FluentCaseInsensitiveStringsMap headers) {
                     hRead.set(headers);
                 }
 
-                public void onBytesReceived(ByteBuffer buffer) {
-                    bbReceivedLenght.addAndGet(buffer.capacity());
+                public void onBytesReceived(byte[] b) {
+                    bbReceivedLenght.addAndGet(b.length);
                 }
 
-                public void onBytesSent(ByteBuffer buffer) {
-                    bbSentLenght.addAndGet(buffer.capacity());
+                public void onBytesSent(long amount, long current, long total) {
+                    bbSentLenght.addAndGet(amount);
                 }
 
                 public void onRequestResponseCompleted() {
@@ -234,19 +250,19 @@ public void onThrowable(Throwable t) {
             });
 
             try {
-                Response response = c.preparePut(getTargetUrl()).setBody(new FileBodyGenerator(largeFile)).execute(tl).get();
+                Response response = client.preparePut(getTargetUrl()).setBody(new FileBodyGenerator(file)).execute(tl).get();
 
                 assertNotNull(response);
                 assertEquals(response.getStatusCode(), 200);
                 assertNotNull(hRead.get());
                 assertNotNull(hSent.get());
-                assertEquals(bbReceivedLenght.get(), largeFile.length());
-                assertEquals(bbSentLenght.get(), largeFile.length());
+                assertEquals(bbReceivedLenght.get(), expectedFileSize, "Number of received bytes incorrect");
+                assertEquals(bbSentLenght.get(), expectedFileSize, "Number of sent bytes incorrect");
             } catch (IOException ex) {
                 fail("Should have timed out");
             }
         } finally {
-            c.close();
+            client.close();
         }
     }
 
@@ -255,20 +271,11 @@ public String getTargetUrl() {
     }
 
     public static File createTempFile(byte[] pattern, int repeat) throws IOException {
-        TMP.mkdirs();
-        TMP.deleteOnExit();
         File tmpFile = File.createTempFile("tmpfile-", ".data", TMP);
-        write(pattern, repeat, tmpFile);
-
-        return tmpFile;
-    }
-
-    public static void write(byte[] pattern, int repeat, File file) throws IOException {
-        file.deleteOnExit();
-        file.getParentFile().mkdirs();
+        tmpFile.deleteOnExit();
         FileOutputStream out = null;
         try {
-            out = new FileOutputStream(file);
+            out = new FileOutputStream(tmpFile);
             for (int i = 0; i < repeat; i++) {
                 out.write(pattern);
             }
@@ -277,5 +284,7 @@ public static void write(byte[] pattern, int repeat, File file) throws IOExcepti
                 out.close();
             }
         }
+
+        return tmpFile;
     }
 }