Have a config parameter for websocket max buffer size, close AsyncHttpClient#658

Author: Stephane Landelle

providers/netty/src/main/java/org/asynchttpclient/providers/netty/NettyAsyncHttpProviderConfig.java

@@ -43,10 +43,8 @@ public class NettyAsyncHttpProviderConfig implements AsyncHttpProviderConfig<Cha
      * Add a property that will be used when the AsyncHttpClient initialize its
      * {@link org.asynchttpclient.AsyncHttpProvider}
      * 
-     * @param name
-     *            the name of the property
-     * @param value
-     *            the value of the property
+     * @param name the name of the property
+     * @param value the value of the property
      * @return this instance of AsyncHttpProviderConfig
      */
     public NettyAsyncHttpProviderConfig addProperty(ChannelOption<Object> name, Object value) {
@@ -116,17 +114,17 @@ public NettyResponseBodyPart newResponseBodyPart(ByteBuf buf, boolean last) {
     }
 
     public static interface NettyWebSocketFactory {
-        NettyWebSocket newNettyWebSocket(Channel channel);
+        NettyWebSocket newNettyWebSocket(Channel channel, NettyAsyncHttpProviderConfig nettyConfig);
     }
 
     public class DefaultNettyWebSocketFactory implements NettyWebSocketFactory {
 
         @Override
-        public NettyWebSocket newNettyWebSocket(Channel channel) {
-            return new NettyWebSocket(channel);
+        public NettyWebSocket newNettyWebSocket(Channel channel, NettyAsyncHttpProviderConfig nettyConfig) {
+            return new NettyWebSocket(channel, nettyConfig);
         }
     }
-    
+
     /**
      * Allow configuring the Netty's event loop.
      */
@@ -166,6 +164,8 @@ public NettyWebSocket newNettyWebSocket(Channel channel) {
 
     private NettyWebSocketFactory nettyWebSocketFactory = new DefaultNettyWebSocketFactory();
 
+    private int webSocketMaxBufferSize = 128000000;
+
     public EventLoopGroup getEventLoopGroup() {
         return eventLoopGroup;
     }
@@ -293,4 +293,12 @@ public NettyWebSocketFactory getNettyWebSocketFactory() {
     public void setNettyWebSocketFactory(NettyWebSocketFactory nettyWebSocketFactory) {
         this.nettyWebSocketFactory = nettyWebSocketFactory;
     }
+
+    public int getWebSocketMaxBufferSize() {
+        return webSocketMaxBufferSize;
+    }
+
+    public void setWebSocketMaxBufferSize(int webSocketMaxBufferSize) {
+        this.webSocketMaxBufferSize = webSocketMaxBufferSize;
+    }
 }

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

@@ -61,7 +61,7 @@ public WebSocketProtocol(ChannelManager channelManager,//
     private void invokeOnSucces(Channel channel, WebSocketUpgradeHandler h) {
         if (!h.touchSuccess()) {
             try {
-                h.onSuccess(nettyConfig.getNettyWebSocketFactory().newNettyWebSocket(channel));
+                h.onSuccess(nettyConfig.getNettyWebSocketFactory().newNettyWebSocket(channel, nettyConfig));
             } catch (Exception ex) {
                 logger.warn("onSuccess unexpected exception", ex);
             }

providers/netty/src/main/java/org/asynchttpclient/providers/netty/ws/NettyWebSocket.java

@@ -31,6 +31,7 @@
 import java.util.concurrent.ConcurrentLinkedQueue;
 
 import org.asynchttpclient.HttpResponseBodyPart;
+import org.asynchttpclient.providers.netty.NettyAsyncHttpProviderConfig;
 import org.asynchttpclient.providers.netty.response.NettyResponseBodyPart;
 import org.asynchttpclient.websocket.WebSocket;
 import org.asynchttpclient.websocket.WebSocketByteFragmentListener;
@@ -50,19 +51,20 @@ public class NettyWebSocket implements WebSocket {
 
     protected final Channel channel;
     protected final Collection<WebSocketListener> listeners;
-    protected int maxBufferSize = 128000000;
+    protected final int maxBufferSize;
     private int bufferSize;
     private List<byte[]> _fragments;
     private volatile boolean interestedInByteMessages;
     private volatile boolean interestedInTextMessages;
 
-    public NettyWebSocket(Channel channel) {
-        this(channel, new ConcurrentLinkedQueue<WebSocketListener>());
+    public NettyWebSocket(Channel channel, NettyAsyncHttpProviderConfig nettyConfig) {
+        this(channel, nettyConfig, new ConcurrentLinkedQueue<WebSocketListener>());
     }
 
-    public NettyWebSocket(Channel channel, Collection<WebSocketListener> listeners) {
+    public NettyWebSocket(Channel channel, NettyAsyncHttpProviderConfig nettyConfig, Collection<WebSocketListener> listeners) {
         this.channel = channel;
         this.listeners = listeners;
+        maxBufferSize = nettyConfig.getWebSocketMaxBufferSize();
     }
 
     @Override
@@ -107,14 +109,6 @@ public WebSocket sendPong(byte[] payload) {
         return this;
     }
 
-    public int getMaxBufferSize() {
-        return maxBufferSize;
-    }
-
-    public void setMaxBufferSize(int maxBufferSize) {
-        this.maxBufferSize = Math.max(maxBufferSize, 8192);
-    }
-
     @Override
     public boolean isOpen() {
         return channel.isOpen();