Revamp WebSocket API, close #1393

Motivation:

Current WebSocket API has many limitations:
1. It's not possible to be notified of received fragmented frames, they
are always aggregated
2. It’s not possible to specify extension bits when sending a frame
3. The API for being notified of write completion is cumbersome
4. There are tons of different interfaces to be notified of the
different types of frames
5. Method names are not aligned between `WebSocket` and
`WebSocketListener`
6. There are 2 onClose listeners, the default one doesn't expose the
code and reason

Modifications:

1. Add a new `aggregateWebSocketFrameFragments` config param,
defaulting to true. When false, fragmented frames are not aggregated.
Drop `WebSocketByteFragmentListener` and
`WebSocketTextFragmentListener` and add `finalFragment` and `rsv`
parameters to `WebSocketListener`
2. Provide a complete set of send methods on `WebSocket` supporting
`rsv` and fragmentation/continuation frame.
3. Have send methods return the Netty future so users can register
listeners
4. Drop `WebSocketTextListener`, `WebSocketByteListener`,
`WebSocketPingListener` and `WebSocketPongListener` and add default
methods on `WebSocketListener`. Drop `DefaultWebSocketListener`.
5. Rename all methods to `sendXXXFrame` and `onXXXFrame`
6. Drop `WebSocketCloseCodeReasonListener` and change
`WebSocketListener#onClose` to notify with code and reason.

Result:

More complete and consistent WebSocket support

Author: slandelle

client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java

@@ -279,6 +279,8 @@ public interface AsyncHttpClientConfig {
 
     boolean isValidateResponseHeaders();
 
+    boolean isAggregateWebSocketFrameFragments();
+
     boolean isTcpNoDelay();
 
     boolean isSoReuseAddress();

client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClientConfig.java

@@ -75,6 +75,7 @@ public class DefaultAsyncHttpClientConfig implements AsyncHttpClientConfig {
     private final boolean keepEncodingHeader;
     private final ProxyServerSelector proxyServerSelector;
     private final boolean validateResponseHeaders;
+    private final boolean aggregateWebSocketFrameFragments;
 
     // timeouts
     private final int connectTimeout;
@@ -148,6 +149,7 @@ private DefaultAsyncHttpClientConfig(//
             boolean keepEncodingHeader,//
             ProxyServerSelector proxyServerSelector,//
             boolean validateResponseHeaders,//
+            boolean aggregateWebSocketFrameFragments,
 
             // timeouts
             int connectTimeout,//
@@ -222,6 +224,7 @@ private DefaultAsyncHttpClientConfig(//
         this.keepEncodingHeader = keepEncodingHeader;
         this.proxyServerSelector = proxyServerSelector;
         this.validateResponseHeaders = validateResponseHeaders;
+        this.aggregateWebSocketFrameFragments = aggregateWebSocketFrameFragments;
 
         // timeouts
         this.connectTimeout = connectTimeout;
@@ -418,6 +421,11 @@ public boolean isValidateResponseHeaders() {
         return validateResponseHeaders;
     }
 
+    @Override
+    public boolean isAggregateWebSocketFrameFragments() {
+        return aggregateWebSocketFrameFragments;
+    }
+
     // ssl
     @Override
     public boolean isUseOpenSsl() {
@@ -617,6 +625,7 @@ public static class Builder {
         private boolean useProxySelector = defaultUseProxySelector();
         private boolean useProxyProperties = defaultUseProxyProperties();
         private boolean validateResponseHeaders = defaultValidateResponseHeaders();
+        private boolean aggregateWebSocketFrameFragments = defaultAggregateWebSocketFrameFragments();
 
         // timeouts
         private int connectTimeout = defaultConnectTimeout();
@@ -819,6 +828,11 @@ public Builder setValidateResponseHeaders(boolean validateResponseHeaders) {
             return this;
         }
 
+        public Builder setAggregateWebSocketFrameFragments(boolean aggregateWebSocketFrameFragments) {
+            this.aggregateWebSocketFrameFragments = aggregateWebSocketFrameFragments;
+            return this;
+        }
+
         public Builder setProxyServer(ProxyServer proxyServer) {
             this.proxyServerSelector = uri -> proxyServer;
             return this;
@@ -1123,6 +1137,7 @@ public DefaultAsyncHttpClientConfig build() {
                     keepEncodingHeader, //
                     resolveProxyServerSelector(), //
                     validateResponseHeaders, //
+                    aggregateWebSocketFrameFragments, //
                     connectTimeout, //
                     requestTimeout, //
                     readTimeout, //

client/src/main/java/org/asynchttpclient/config/AsyncHttpClientConfigDefaults.java

@@ -98,6 +98,10 @@ public static boolean defaultValidateResponseHeaders() {
         return AsyncHttpClientConfigHelper.getAsyncHttpClientConfig().getBoolean(ASYNC_CLIENT_CONFIG_ROOT + "validateResponseHeaders");
     }
 
+    public static boolean defaultAggregateWebSocketFrameFragments() {
+        return AsyncHttpClientConfigHelper.getAsyncHttpClientConfig().getBoolean(ASYNC_CLIENT_CONFIG_ROOT + "aggregateWebSocketFrameFragments");
+    }
+
     public static boolean defaultStrict302Handling() {
         return AsyncHttpClientConfigHelper.getAsyncHttpClientConfig().getBoolean(ASYNC_CLIENT_CONFIG_ROOT + "strict302Handling");
     }

client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java

@@ -396,7 +396,9 @@ public Bootstrap getBootstrap(Uri uri, ProxyServer proxy) {
     public void upgradePipelineForWebSockets(ChannelPipeline pipeline) {
         pipeline.addAfter(HTTP_CLIENT_CODEC, WS_ENCODER_HANDLER, new WebSocket08FrameEncoder(true));
         pipeline.addBefore(AHC_WS_HANDLER, WS_DECODER_HANDLER, new WebSocket08FrameDecoder(false, false, config.getWebSocketMaxFrameSize()));
-        pipeline.addAfter(WS_DECODER_HANDLER, WS_FRAME_AGGREGATOR, new WebSocketFrameAggregator(config.getWebSocketMaxBufferSize()));
+        if (config.isAggregateWebSocketFrameFragments()) {
+            pipeline.addAfter(WS_DECODER_HANDLER, WS_FRAME_AGGREGATOR, new WebSocketFrameAggregator(config.getWebSocketMaxBufferSize()));
+        }
         pipeline.remove(HTTP_CLIENT_CODEC);
     }
 

client/src/main/java/org/asynchttpclient/netty/handler/WebSocketHandler.java

@@ -91,6 +91,14 @@ private void abort(Channel channel, NettyResponseFuture<?> future, WebSocketUpgr
         }
     }
 
+    private static WebSocketUpgradeHandler getWebSocketUpgradeHandler(NettyResponseFuture<?> future) {
+        return (WebSocketUpgradeHandler) future.getAsyncHandler();
+    }
+    
+    private static NettyWebSocket getNettyWebSocket(NettyResponseFuture<?> future) throws Exception {
+        return getWebSocketUpgradeHandler(future).onCompleted();
+    }
+    
     @Override
     public void handleRead(Channel channel, NettyResponseFuture<?> future, Object e) throws Exception {
 
@@ -101,7 +109,7 @@ public void handleRead(Channel channel, NettyResponseFuture<?> future, Object e)
                 logger.debug("\n\nRequest {}\n\nResponse {}\n", httpRequest, response);
             }
 
-            WebSocketUpgradeHandler handler = (WebSocketUpgradeHandler) future.getAsyncHandler();
+            WebSocketUpgradeHandler handler = getWebSocketUpgradeHandler(future);
             HttpResponseStatus status = new NettyResponseStatus(future.getUri(), response, channel);
             HttpResponseHeaders responseHeaders = new HttpResponseHeaders(response.headers());
 
@@ -116,9 +124,8 @@ public void handleRead(Channel channel, NettyResponseFuture<?> future, Object e)
             }
 
         } else if (e instanceof WebSocketFrame) {
-            final WebSocketFrame frame = (WebSocketFrame) e;
-            WebSocketUpgradeHandler handler = (WebSocketUpgradeHandler) future.getAsyncHandler();
-            NettyWebSocket webSocket = handler.onCompleted();
+            WebSocketFrame frame = (WebSocketFrame) e;
+            NettyWebSocket webSocket = getNettyWebSocket(future);
             // retain because we might buffer the frame
             if (webSocket.isReady()) {
                 webSocket.handleFrame(frame);
@@ -139,11 +146,10 @@ public void handleException(NettyResponseFuture<?> future, Throwable e) {
         logger.warn("onError", e);
 
         try {
-            WebSocketUpgradeHandler h = (WebSocketUpgradeHandler) future.getAsyncHandler();
-            NettyWebSocket webSocket = h.onCompleted();
+            NettyWebSocket webSocket = getNettyWebSocket(future);
             if (webSocket != null) {
                 webSocket.onError(e.getCause());
-                webSocket.close();
+                webSocket.sendCloseFrame();
             }
         } catch (Throwable t) {
             logger.error("onError", t);
@@ -152,15 +158,13 @@ public void handleException(NettyResponseFuture<?> future, Throwable e) {
 
     @Override
     public void handleChannelInactive(NettyResponseFuture<?> future) {
-        logger.trace("onClose");
+        logger.trace("Connection was closed abnormally (that is, with no close frame being received).");
 
         try {
-            WebSocketUpgradeHandler h = (WebSocketUpgradeHandler) future.getAsyncHandler();
-            NettyWebSocket webSocket = h.onCompleted();
-
-            logger.trace("Connection was closed abnormally (that is, with no close frame being received).");
-            if (webSocket != null)
-                webSocket.close(1006, "Connection was closed abnormally (that is, with no close frame being received).");
+            NettyWebSocket webSocket = getNettyWebSocket(future);
+            if (webSocket != null) {
+                webSocket.onClose(1006, "Connection was closed abnormally (that is, with no close frame being received).");
+            }
         } catch (Throwable t) {
             logger.error("onError", t);
         }

client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java

@@ -339,8 +339,7 @@ public <T> void writeRequest(NettyResponseFuture<T> future, Channel channel) {
         HttpRequest httpRequest = nettyRequest.getHttpRequest();
         AsyncHandler<T> handler = future.getAsyncHandler();
 
-        // if the channel is dead because it was pooled and the remote
-        // server decided to close it,
+        // if the channel is dead because it was pooled and the remote server decided to close it,
         // we just let it go and the channelInactive do its work
         if (!Channels.isChannelValid(channel))
             return;
@@ -366,6 +365,7 @@ public <T> void writeRequest(NettyResponseFuture<T> future, Channel channel) {
 
                 // if the request has a body, we want to track progress
                 if (writeBody) {
+                    // FIXME does this really work??? the promise is for the request without body!!!
                     ChannelProgressivePromise promise = channel.newProgressivePromise();
                     ChannelFuture f = channel.write(httpRequest, promise);
                     f.addListener(new WriteProgressListener(future, true, 0L));