Share applyResponseFiltersAndReplayRequest method for WS too

Author: Stephane Landelle

api/src/main/java/org/asynchttpclient/AsyncHttpClient.java

@@ -581,20 +581,16 @@ public ListenableFuture<Response> executeRequest(Request request) throws IOExcep
      * @return {@link FilterContext}
      */
     private <T> FilterContext<T> preProcessRequest(FilterContext<T> fc) throws IOException {
-        if (config.hasRequestFilters()) {
-            final List<RequestFilter> requestFilters = config.getRequestFilters();
-            for (int i = 0, len = requestFilters.size(); i < len; i++) {
-                final RequestFilter asyncFilter = requestFilters.get(i);
-                try {
-                    fc = asyncFilter.filter(fc);
-                    if (fc == null) {
-                        throw new NullPointerException("FilterContext is null");
-                    }
-                } catch (FilterException e) {
-                    IOException ex = new IOException();
-                    ex.initCause(e);
-                    throw ex;
+        for (RequestFilter asyncFilter: config.getRequestFilters()) {
+            try {
+                fc = asyncFilter.filter(fc);
+                if (fc == null) {
+                    throw new NullPointerException("FilterContext is null");
                 }
+            } catch (FilterException e) {
+                IOException ex = new IOException();
+                ex.initCause(e);
+                throw ex;
             }
         }
 

providers/netty/src/main/java/org/asynchttpclient/providers/netty/handler/HttpProtocol.java

@@ -63,7 +63,7 @@
 import org.slf4j.LoggerFactory;
 
 final class HttpProtocol extends Protocol {
-    
+
     private static final Logger LOGGER = LoggerFactory.getLogger(HttpProtocol.class);
 
     public HttpProtocol(Channels channels, AsyncHttpClientConfig config, NettyAsyncHttpProviderConfig nettyConfig, NettyRequestSender requestSender) {
@@ -212,36 +212,6 @@ private void markAsDone(NettyResponseFuture<?> future, final ChannelHandlerConte
         }
     }
 
-    private boolean applyResponseFiltersAndReplayRequest(ChannelHandlerContext ctx, NettyResponseFuture<?> future, HttpResponseStatus status, HttpResponseHeaders responseHeaders)
-            throws IOException {
-
-        AsyncHandler handler = future.getAsyncHandler();
-        FilterContext fc = new FilterContext.FilterContextBuilder().asyncHandler(handler).request(future.getRequest()).responseStatus(status).responseHeaders(responseHeaders)
-                .build();
-
-        for (ResponseFilter asyncFilter : config.getResponseFilters()) {
-            try {
-                fc = asyncFilter.filter(fc);
-                // FIXME Is it worth protecting against this?
-                if (fc == null) {
-                    throw new NullPointerException("FilterContext is null");
-                }
-            } catch (FilterException efe) {
-                channels.abort(future, efe);
-            }
-        }
-
-        // The handler may have been wrapped.
-        future.setAsyncHandler(fc.getAsyncHandler());
-
-        // The request has changed
-        if (fc.replayRequest()) {
-            requestSender.replayRequest(future, fc, ctx);
-            return true;
-        }
-        return false;
-    }
-
     private boolean handleResponseAndExit(final ChannelHandlerContext ctx, final NettyResponseFuture<?> future, AsyncHandler<?> handler, HttpRequest nettyRequest,
             ProxyServer proxyServer, HttpResponse response) throws Exception {
         Request request = future.getRequest();
@@ -258,7 +228,7 @@ private boolean handleResponseAndExit(final ChannelHandlerContext ctx, final Net
 
         future.setKeepAlive(!HttpHeaders.Values.CLOSE.equalsIgnoreCase(response.headers().get(HttpHeaders.Names.CONNECTION)));
 
-        if (!config.getResponseFilters().isEmpty() && applyResponseFiltersAndReplayRequest(ctx, future, status, responseHeaders)) {
+        if (applyResponseFiltersAndReplayRequest(ctx, future, status, responseHeaders)) {
             return true;
         }
 

providers/netty/src/main/java/org/asynchttpclient/providers/netty/handler/Protocol.java

@@ -19,13 +19,20 @@
 import io.netty.handler.codec.http.HttpMethod;
 import io.netty.handler.codec.http.HttpResponse;
 
+import java.io.IOException;
 import java.net.URI;
 
+import org.asynchttpclient.AsyncHandler;
 import org.asynchttpclient.AsyncHttpClientConfig;
 import org.asynchttpclient.Cookie;
+import org.asynchttpclient.HttpResponseHeaders;
+import org.asynchttpclient.HttpResponseStatus;
 import org.asynchttpclient.MaxRedirectException;
 import org.asynchttpclient.Request;
 import org.asynchttpclient.RequestBuilder;
+import org.asynchttpclient.filter.FilterContext;
+import org.asynchttpclient.filter.FilterException;
+import org.asynchttpclient.filter.ResponseFilter;
 import org.asynchttpclient.org.jboss.netty.handler.codec.http.CookieDecoder;
 import org.asynchttpclient.providers.netty.Callback;
 import org.asynchttpclient.providers.netty.NettyAsyncHttpProviderConfig;
@@ -63,9 +70,12 @@ protected boolean redirect(Request request, NettyResponseFuture<?> future, HttpR
         io.netty.handler.codec.http.HttpResponseStatus status = response.getStatus();
         boolean redirectEnabled = request.isRedirectOverrideSet() ? request.isRedirectEnabled() : config.isRedirectEnabled();
         boolean isRedirectStatus = status.equals(MOVED_PERMANENTLY) || status.equals(FOUND) || status.equals(SEE_OTHER) || status.equals(TEMPORARY_REDIRECT);
+
         if (redirectEnabled && isRedirectStatus) {
+            if (future.incrementAndGetCurrentRedirectCount() >= config.getMaxRedirects()) {
+                throw new MaxRedirectException("Maximum redirect reached: " + config.getMaxRedirects());
 
-            if (future.incrementAndGetCurrentRedirectCount() < config.getMaxRedirects()) {
+            } else {
                 // We must allow 401 handling again.
                 future.getAndSetAuth(false);
 
@@ -132,8 +142,38 @@ public void call() throws Exception {
                     requestSender.sendNextRequest(target, future);
                     return true;
                 }
-            } else {
-                throw new MaxRedirectException("Maximum redirect reached: " + config.getMaxRedirects());
+            }
+        }
+        return false;
+    }
+
+    protected boolean applyResponseFiltersAndReplayRequest(ChannelHandlerContext ctx, NettyResponseFuture<?> future, HttpResponseStatus status, HttpResponseHeaders responseHeaders)
+            throws IOException {
+
+        if (!config.getResponseFilters().isEmpty()) {
+            AsyncHandler handler = future.getAsyncHandler();
+            FilterContext fc = new FilterContext.FilterContextBuilder().asyncHandler(handler).request(future.getRequest()).responseStatus(status).responseHeaders(responseHeaders)
+                    .build();
+
+            for (ResponseFilter asyncFilter : config.getResponseFilters()) {
+                try {
+                    fc = asyncFilter.filter(fc);
+                    // FIXME Is it worth protecting against this?
+                    if (fc == null) {
+                        throw new NullPointerException("FilterContext is null");
+                    }
+                } catch (FilterException efe) {
+                    channels.abort(future, efe);
+                }
+            }
+
+            // The handler may have been wrapped.
+            future.setAsyncHandler(fc.getAsyncHandler());
+
+            // The request has changed
+            if (fc.replayRequest()) {
+                requestSender.replayRequest(future, fc, ctx);
+                return true;
             }
         }
         return false;

providers/netty/src/main/java/org/asynchttpclient/providers/netty/handler/WebSocketProtocol.java

@@ -27,6 +27,7 @@
 import io.netty.handler.codec.http.websocketx.WebSocketFrame;
 
 import java.io.IOException;
+import java.util.Locale;
 
 import org.asynchttpclient.AsyncHandler.STATE;
 import org.asynchttpclient.AsyncHttpClientConfig;
@@ -84,29 +85,10 @@ public void handle(ChannelHandlerContext ctx, NettyResponseFuture future, Object
 
         if (e instanceof HttpResponse) {
             HttpResponse response = (HttpResponse) e;
-
-            HttpResponseStatus s = new ResponseStatus(future.getURI(), response, config);
+            HttpResponseStatus status = new ResponseStatus(future.getURI(), response, config);
             HttpResponseHeaders responseHeaders = new ResponseHeaders(future.getURI(), response.headers());
 
-            // FIXME there's a method for that IIRC
-            FilterContext<?> fc = new FilterContext.FilterContextBuilder().asyncHandler(h).request(request).responseStatus(s).responseHeaders(responseHeaders).build();
-            for (ResponseFilter asyncFilter : config.getResponseFilters()) {
-                try {
-                    fc = asyncFilter.filter(fc);
-                    if (fc == null) {
-                        throw new NullPointerException("FilterContext is null");
-                    }
-                } catch (FilterException efe) {
-                    channels.abort(future, efe);
-                }
-            }
-
-            // The handler may have been wrapped.
-            future.setAsyncHandler(fc.getAsyncHandler());
-
-            // The request has changed
-            if (fc.replayRequest()) {
-                requestSender.replayRequest(future, fc, ctx);
+            if (applyResponseFiltersAndReplayRequest(ctx, future, status, responseHeaders)) {
                 return;
             }
 
@@ -118,13 +100,13 @@ public void handle(ChannelHandlerContext ctx, NettyResponseFuture future, Object
             boolean validUpgrade = response.headers().get(HttpHeaders.Names.UPGRADE) != null;
             String c = response.headers().get(HttpHeaders.Names.CONNECTION);
             if (c == null) {
-                c = response.headers().get(HttpHeaders.Names.CONNECTION.toLowerCase());
+                c = response.headers().get(HttpHeaders.Names.CONNECTION.toLowerCase(Locale.ENGLISH));
             }
 
-            boolean validConnection = c == null ? false : c.equalsIgnoreCase(HttpHeaders.Values.UPGRADE);
+            boolean validConnection = c != null && c.equalsIgnoreCase(HttpHeaders.Values.UPGRADE);
 
-            s = new ResponseStatus(future.getURI(), response, config);
-            final boolean statusReceived = h.onStatusReceived(s) == STATE.UPGRADE;
+            status = new ResponseStatus(future.getURI(), response, config);
+            final boolean statusReceived = h.onStatusReceived(status) == STATE.UPGRADE;
 
             final boolean headerOK = h.onHeadersReceived(responseHeaders) == STATE.CONTINUE;
             if (!headerOK || !validStatus || !validUpgrade || !validConnection || !statusReceived) {
@@ -222,12 +204,11 @@ public void onClose(ChannelHandlerContext ctx) {
             WebSocketUpgradeHandler h = WebSocketUpgradeHandler.class.cast(nettyResponse.getAsyncHandler());
             NettyWebSocket webSocket = NettyWebSocket.class.cast(h.onCompleted());
 
-            // FIXME How could this test not succeed, attachment is a
-            // NettyResponseFuture????
+            // FIXME How could this test not succeed, we just checked above that attribute is a NettyResponseFuture????
             if (attribute != DiscardEvent.INSTANCE)
                 webSocket.close(1006, "Connection was closed abnormally (that is, with no close frame being sent).");
         } catch (Throwable t) {
             LOGGER.error("onError", t);
         }
     }
-}
+}