switch to 4.1.0.CR1-SNAPSHOT, see netty/netty#4593

Author: slandelle

client/pom.xml

@@ -28,12 +28,12 @@
 		<dependency>
 			<groupId>io.netty</groupId>
 			<artifactId>netty-codec-http</artifactId>
-			<version>4.1.0.Beta8</version>
+			<version>4.1.0.CR1-SNAPSHOT</version>
 		</dependency>
 		<dependency>
 			<groupId>io.netty</groupId>
 			<artifactId>netty-handler</artifactId>
-			<version>4.1.0.Beta8</version>
+			<version>4.1.0.CR1-SNAPSHOT</version>
 		</dependency>
 		<dependency>
 			<groupId>org.reactivestreams</groupId>
@@ -43,7 +43,7 @@
 		<dependency>
 			<groupId>com.typesafe.netty</groupId>
 			<artifactId>netty-reactive-streams</artifactId>
-			<version>1.0.0</version>
+			<version>1.0.1</version>
 			<exclusions>
 				<exclusion>
 					<groupId>io.netty</groupId>

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

@@ -398,10 +398,13 @@ public static boolean isSslHandlerConfigured(ChannelPipeline pipeline) {
     }
 
     public void upgradeProtocol(ChannelPipeline pipeline, Uri requestUri) throws SSLException {
-        if (pipeline.get("HttpClientCodec$Decoder#0") != null) {
-            pipeline.remove("HttpClientCodec$Decoder#0");
-            pipeline.remove("HttpClientCodec$Encoder#0");
-        }
+
+        if (pipeline.get(HTTP_CLIENT_CODEC) != null)
+            pipeline.remove(HTTP_CLIENT_CODEC);
+        // if (pipeline.get("HttpClientCodec$Decoder#0") != null) {
+        // pipeline.remove("HttpClientCodec$Decoder#0");
+        // pipeline.remove("HttpClientCodec$Encoder#0");
+        // }
 
         if (requestUri.isSecured())
             if (isSslHandlerConfigured(pipeline)) {
@@ -450,12 +453,14 @@ public Bootstrap getBootstrap(Uri uri, ProxyServer proxy) {
 
     public void upgradePipelineForWebSockets(ChannelPipeline pipeline) {
 
-//        HttpClientCodec$Decoder#0
-//        HttpClientCodec$Encoder#0
-        
-        pipeline.addAfter("HttpClientCodec$Encoder#0", WS_ENCODER_HANDLER, new WebSocket08FrameEncoder(true));
-        pipeline.remove("HttpClientCodec$Decoder#0");
-        pipeline.remove("HttpClientCodec$Encoder#0");
+//        pipeline.addAfter(HTTP_CLIENT_CODEC, WS_ENCODER_HANDLER, new WebSocket08FrameEncoder(true));
+        pipeline.addBefore(AHC_WS_HANDLER, WS_ENCODER_HANDLER, new WebSocket08FrameEncoder(true));
+        pipeline.remove(HTTP_CLIENT_CODEC);
+        // HttpClientCodec$Decoder#0
+        // HttpClientCodec$Encoder#0
+        // pipeline.addAfter("HttpClientCodec$Encoder#0", WS_ENCODER_HANDLER, new WebSocket08FrameEncoder(true));
+        // pipeline.remove("HttpClientCodec$Decoder#0");
+        // pipeline.remove("HttpClientCodec$Encoder#0");
         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()));
     }

client/src/main/java/org/asynchttpclient/netty/request/body/BodyChunkedInput.java

@@ -15,6 +15,7 @@
 
 import static org.asynchttpclient.util.Assertions.assertNotNull;
 import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.stream.ChunkedInput;
 
@@ -45,24 +46,28 @@ public BodyChunkedInput(Body body) {
 
     @Override
     public ByteBuf readChunk(ChannelHandlerContext ctx) throws Exception {
+        return readChunk(ctx.alloc());
+    }
 
+    @Override
+    public ByteBuf readChunk(ByteBufAllocator allocator) throws Exception {
         if (endOfInput)
             return null;
 
-        ByteBuf buffer = ctx.alloc().buffer(chunkSize);
+        ByteBuf buffer = allocator.buffer(chunkSize);
         Body.BodyState state = body.transferTo(buffer);
         progress += buffer.readableBytes();
         switch (state) {
-            case STOP:
-                endOfInput = true;
-                return buffer;
-            case SUSPEND:
-                //this will suspend the stream in ChunkedWriteHandler
-                return null;
-            case CONTINUE:
-                return buffer;
-            default:
-                throw new IllegalStateException("Unknown state: " + state);
+        case STOP:
+            endOfInput = true;
+            return buffer;
+        case SUSPEND:
+            // this will suspend the stream in ChunkedWriteHandler
+            return null;
+        case CONTINUE:
+            return buffer;
+        default:
+            throw new IllegalStateException("Unknown state: " + state);
         }
     }
 
@@ -75,12 +80,12 @@ public boolean isEndOfInput() throws Exception {
     public void close() throws Exception {
         body.close();
     }
-    
+
     @Override
     public long progress() {
         return progress;
     }
-    
+
     @Override
     public long length() {
         return length;