no need to make lastProgress an AtomicLong, Listeners run in the Channel Thread

Author: Stephane Landelle

providers/netty/src/main/java/org/asynchttpclient/providers/netty/request/ProgressListener.java

@@ -15,11 +15,11 @@
  */
 package org.asynchttpclient.providers.netty.request;
 
+import io.netty.channel.Channel;
 import io.netty.channel.ChannelProgressiveFuture;
 import io.netty.channel.ChannelProgressiveFutureListener;
 
 import java.nio.channels.ClosedChannelException;
-import java.util.concurrent.atomic.AtomicLong;
 
 import org.asynchttpclient.AsyncHandler;
 import org.asynchttpclient.AsyncHttpClientConfig;
@@ -31,14 +31,14 @@
 import org.slf4j.LoggerFactory;
 
 public class ProgressListener implements ChannelProgressiveFutureListener {
-    
+
     private static final Logger LOGGER = LoggerFactory.getLogger(ProgressListener.class);
 
     private final AsyncHttpClientConfig config;
     private final boolean notifyHeaders;
     private final AsyncHandler<?> asyncHandler;
     private final NettyResponseFuture<?> future;
-    private final AtomicLong lastProgress = new AtomicLong(0);
+    private long lastProgress = 0L;
 
     public ProgressListener(AsyncHttpClientConfig config, boolean notifyHeaders, AsyncHandler<?> asyncHandler, NettyResponseFuture<?> future) {
         this.config = config;
@@ -47,54 +47,60 @@ public ProgressListener(AsyncHttpClientConfig config, boolean notifyHeaders, Asy
         this.future = future;
     }
 
-    @Override
-    public void operationComplete(ChannelProgressiveFuture cf) {
-        // The write operation failed. If the channel was cached, it means it got asynchronously closed.
-        // Let's retry a second time.
-        Throwable cause = cf.cause();
+    private boolean abortOnThrowable(Throwable cause, Channel channel) {
+
         if (cause != null && future.getState() != NettyResponseFuture.STATE.NEW) {
 
             if (cause instanceof IllegalStateException) {
                 LOGGER.debug(cause.getMessage(), cause);
                 try {
-                    cf.channel().close();
+                    channel.close();
                 } catch (RuntimeException ex) {
                     LOGGER.debug(ex.getMessage(), ex);
                 }
-                return;
-            }
-
-            if (cause instanceof ClosedChannelException || NettyResponseFutures.abortOnReadCloseException(cause) || NettyResponseFutures.abortOnWriteCloseException(cause)) {
+            } else if (cause instanceof ClosedChannelException || NettyResponseFutures.abortOnReadCloseException(cause) || NettyResponseFutures.abortOnWriteCloseException(cause)) {
 
-                if (LOGGER.isDebugEnabled()) {
-                    LOGGER.debug(cf.cause() == null ? "" : cf.cause().getMessage(), cf.cause());
-                }
+                if (LOGGER.isDebugEnabled())
+                    LOGGER.debug(cause.getMessage(), cause);
 
                 try {
-                    cf.channel().close();
+                    channel.close();
                 } catch (RuntimeException ex) {
                     LOGGER.debug(ex.getMessage(), ex);
                 }
-                return;
             } else {
                 future.abort(cause);
             }
-            return;
+
+            return true;
         }
-        future.touch();
 
-        /**
-         * We need to make sure we aren't in the middle of an authorization process before publishing events as we will re-publish again the same event after the authorization,
-         * causing unpredictable behavior.
-         */
-        Realm realm = future.getRequest().getRealm() != null ? future.getRequest().getRealm() : config.getRealm();
-        boolean startPublishing = future.isInAuth() || realm == null || realm.getUsePreemptiveAuth();
+        return false;
+    }
 
-        if (startPublishing && asyncHandler instanceof ProgressAsyncHandler) {
-            if (notifyHeaders) {
-                ProgressAsyncHandler.class.cast(asyncHandler).onHeaderWriteCompleted();
-            } else {
-                ProgressAsyncHandler.class.cast(asyncHandler).onContentWriteCompleted();
+    @Override
+    public void operationComplete(ChannelProgressiveFuture cf) {
+        // The write operation failed. If the channel was cached, it means it got asynchronously closed.
+        // Let's retry a second time.
+
+        if (!abortOnThrowable(cf.cause(), cf.channel())) {
+
+            future.touch();
+
+            /**
+             * We need to make sure we aren't in the middle of an authorization process before publishing events as we will re-publish again the same event after the authorization,
+             * causing unpredictable behavior.
+             */
+            Realm realm = future.getRequest().getRealm() != null ? future.getRequest().getRealm() : config.getRealm();
+            boolean startPublishing = future.isInAuth() || realm == null || realm.getUsePreemptiveAuth();
+
+            if (startPublishing && asyncHandler instanceof ProgressAsyncHandler) {
+                ProgressAsyncHandler<?> progressAsyncHandler = (ProgressAsyncHandler<?>) asyncHandler;
+                if (notifyHeaders) {
+                    progressAsyncHandler.onHeaderWriteCompleted();
+                } else {
+                    progressAsyncHandler.onContentWriteCompleted();
+                }
             }
         }
     }
@@ -103,8 +109,9 @@ public void operationComplete(ChannelProgressiveFuture cf) {
     public void operationProgressed(ChannelProgressiveFuture f, long progress, long total) {
         future.touch();
         if (!notifyHeaders && asyncHandler instanceof ProgressAsyncHandler) {
-            long lastProgressValue = lastProgress.getAndSet(progress);
-            ProgressAsyncHandler.class.cast(asyncHandler).onContentWriteProgress(progress - lastProgressValue, progress, total);
+            long lastLastProgress = lastProgress;
+            lastProgress = progress;
+            ProgressAsyncHandler.class.cast(asyncHandler).onContentWriteProgress(progress - lastLastProgress, progress, total);
         }
     }
 }