Improve TransferListener supports by making sure no events are propagated if a request retry occurs

Author: jfarcand

src/main/java/com/ning/http/client/listener/TransferCompletionHandler.java

@@ -190,7 +190,7 @@ public void onThrowable(Throwable t) {
     private void fireOnHeadersSent(FluentCaseInsensitiveStringsMap headers){
         for (TransferListener l: listeners) {
             try {
-                l.onResponseHeadersReceived(headers);
+                l.onRequestHeadersSent(headers);
             } catch (Throwable t) {
                 l.onThrowable(t);
             }
@@ -200,7 +200,7 @@ private void fireOnHeadersSent(FluentCaseInsensitiveStringsMap headers){
     private void fireOnHeaderReceived(FluentCaseInsensitiveStringsMap headers){
         for (TransferListener l: listeners) {
             try {
-                l.onRequestHeadersSent(headers);
+                l.onResponseHeadersReceived(headers);
             } catch (Throwable t) {
                 l.onThrowable(t);
             }

src/main/java/com/ning/http/client/providers/netty/NettyAsyncHttpProvider.java

@@ -1578,7 +1578,7 @@ public static <T> NettyResponseFuture<T> newFuture(URI uri,
         return f;
     }
 
-    private static class ProgressListener implements ChannelFutureProgressListener {
+    private class ProgressListener implements ChannelFutureProgressListener {
 
         private final boolean notifyHeaders;
         private final AsyncHandler asyncHandler;
@@ -1625,7 +1625,12 @@ public void operationComplete(ChannelFuture cf) {
             }
             future.touch();
 
-            if (ProgressAsyncHandler.class.isAssignableFrom(asyncHandler.getClass())) {
+            Realm realm = future.getRequest().getRealm() != null ? future.getRequest().getRealm() : NettyAsyncHttpProvider.this.getConfig().getRealm();
+            boolean startPublishing = future.isInAuth()
+                    || realm == null
+                    || realm.getUsePreemptiveAuth() == true;
+
+            if (startPublishing && ProgressAsyncHandler.class.isAssignableFrom(asyncHandler.getClass())) {
                 if (notifyHeaders) {
                     ProgressAsyncHandler.class.cast(asyncHandler).onHeaderWriteCompleted();
                 } else {

src/test/java/com/ning/http/client/async/TransferListenerTest.java

@@ -15,6 +15,7 @@
 import com.ning.http.client.AsyncHttpClient;
 import com.ning.http.client.FluentCaseInsensitiveStringsMap;
 import com.ning.http.client.Response;
+import com.ning.http.client.generators.FileBodyGenerator;
 import com.ning.http.client.listener.TransferCompletionHandler;
 import com.ning.http.client.listener.TransferListener;
 import org.eclipse.jetty.server.handler.AbstractHandler;
@@ -190,6 +191,66 @@ public void onThrowable(Throwable t) {
         c.close();
     }
 
+    @Test(groups = {"standalone", "default_provider"})
+    public void basicPutBodyTest() throws Throwable {
+        AsyncHttpClient c = new AsyncHttpClient();
+
+        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 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);
+
+        TransferCompletionHandler tl = new TransferCompletionHandler();
+        tl.addTransferListener(new TransferListener() {
+
+            public void onRequestHeadersSent(FluentCaseInsensitiveStringsMap headers) {
+                hSent.set(headers);
+            }
+
+            public void onResponseHeadersReceived(FluentCaseInsensitiveStringsMap headers) {
+                hRead.set(headers);
+            }
+
+            public void onBytesReceived(ByteBuffer buffer) {
+                bbReceivedLenght.addAndGet(buffer.capacity());
+            }
+
+            public void onBytesSent(ByteBuffer buffer) {
+                bbSentLenght.addAndGet(buffer.capacity());
+            }
+
+            public void onRequestResponseCompleted() {
+                completed.set(true);
+            }
+
+            public void onThrowable(Throwable t) {
+                throwable.set(t);
+            }
+        });
+
+        try {
+            Response response = c.preparePut(getTargetUrl()).setBody(new FileBodyGenerator(largeFile))
+                    .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());
+        } catch (IOException ex) {
+            fail("Should have timed out");
+        }
+        c.close();
+    }
+
     public String getTargetUrl() {
         return String.format("http://127.0.0.1:%d/foo/test", port1);
     }