Notify of WebSocket Continuation frames when aggregateWebSocketFrameFragments is disabled, close #1396

Motivation:

98bef40 introduced a aggregateWebSocketFrameFragments config option.
But when it's set to false, WebSocketListener are not notified with
those.

Modifications:

Memorize current fragment type and notify of Continuation frames
content.

Result:

Fixed support of non aggregated fragmented frames

Author: slandelle

Diff for: client/src/main/java/org/asynchttpclient/netty/ws/NettyWebSocket.java

@@ -52,6 +52,7 @@ public class NettyWebSocket implements WebSocket {
     // no need for volatile because only mutated in IO thread
     private boolean ready;
     private List<WebSocketFrame> bufferedFrames;
+    protected FragmentedFrameType expectedFragmentedFrameType;
 
     public NettyWebSocket(Channel channel, HttpHeaders upgradeHeaders) {
         this(channel, upgradeHeaders, new ConcurrentLinkedQueue<>());
@@ -238,6 +239,9 @@ public void handleFrame(WebSocketFrame frame) {
 
         } else if (frame instanceof PongWebSocketFrame) {
             onPongFrame((PongWebSocketFrame) frame);
+
+        } else if (frame instanceof ContinuationWebSocketFrame) {
+            onContinuationFrame((ContinuationWebSocketFrame) frame);
         }
     }
 
@@ -276,13 +280,27 @@ public String toString() {
     }
 
     public void onBinaryFrame(BinaryWebSocketFrame frame) {
+        if (expectedFragmentedFrameType == null && !frame.isFinalFragment()) {
+            expectedFragmentedFrameType = FragmentedFrameType.BINARY;
+        }
+        onBinaryFrame0(frame);
+    }
+
+    private void onBinaryFrame0(WebSocketFrame frame) {
         byte[] bytes = byteBuf2Bytes(frame.content());
         for (WebSocketListener listener : listeners) {
             listener.onBinaryFrame(bytes, frame.isFinalFragment(), frame.rsv());
         }
     }
 
     public void onTextFrame(TextWebSocketFrame frame) {
+        if (expectedFragmentedFrameType == null && !frame.isFinalFragment()) {
+            expectedFragmentedFrameType = FragmentedFrameType.TEXT;
+        }
+        onTextFrame0(frame);
+    }
+
+    private void onTextFrame0(WebSocketFrame frame) {
         try {
             // faster than frame.text();
             String text = Utf8ByteBufCharsetDecoder.decodeUtf8(frame.content());
@@ -296,6 +314,29 @@ public void onTextFrame(TextWebSocketFrame frame) {
         }
     }
 
+    public void onContinuationFrame(ContinuationWebSocketFrame frame) {
+        if (expectedFragmentedFrameType == null) {
+            LOGGER.warn("Received continuation frame without an original text or binary frame, ignoring");
+            return;
+        }
+        try {
+            switch (expectedFragmentedFrameType) {
+            case BINARY:
+                onBinaryFrame0(frame);
+                break;
+            case TEXT:
+                onTextFrame0(frame);
+                break;
+            default:
+                throw new IllegalArgumentException("Unknown FragmentedFrameType " + expectedFragmentedFrameType);
+            }
+        } finally {
+            if (frame.isFinalFragment()) {
+                expectedFragmentedFrameType = null;
+            }
+        }
+    }
+
     public void onPingFrame(PingWebSocketFrame frame) {
         byte[] bytes = byteBuf2Bytes(frame.content());
         for (WebSocketListener listener : listeners) {
@@ -309,4 +350,8 @@ public void onPongFrame(PongWebSocketFrame frame) {
             listener.onPongFrame(bytes);
         }
     }
+
+    private enum FragmentedFrameType {
+        TEXT, BINARY;
+    }
 }