Port dcb6156 on master, close AsyncHttpClient#301

Author: Stephane Landelle

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

@@ -37,8 +37,8 @@ public class NettyWebSocket implements WebSocket {
  private final Channel channel;
  private final ConcurrentLinkedQueue<WebSocketListener> listeners = new ConcurrentLinkedQueue<WebSocketListener>();
 
- private StringBuilder textBuffer;
- private ByteArrayOutputStream byteBuffer;
+ private final StringBuilder textBuffer = new StringBuilder();
+ private final ByteArrayOutputStream byteBuffer = new ByteArrayOutputStream();
  private int maxBufferSize = 128000000;
 
  public NettyWebSocket(Channel channel) {
@@ -136,65 +136,87 @@ public void close(int statusCode, String reason) {
  }
 
  public void onBinaryFragment(byte[] message, boolean last) {
- for (WebSocketListener l : listeners) {
- if (l instanceof WebSocketByteListener) {
- try {
- WebSocketByteListener.class.cast(l).onFragment(message, last);
 
- if (byteBuffer == null) {
- byteBuffer = new ByteArrayOutputStream();
- }
-
- byteBuffer.write(message);
+ if (!last) {
+ try {
+ byteBuffer.write(message);
+ } catch (Exception ex) {
+ byteBuffer.reset();
+ onError(ex);
+ return;
+ }
 
- if (byteBuffer.size() > maxBufferSize) {
- Exception e = new Exception("Exceeded Netty Web Socket maximum buffer size of " + getMaxBufferSize());
- l.onError(e);
- this.close();
- return;
- }
+ if (byteBuffer.size() > maxBufferSize) {
+ byteBuffer.reset();
+ Exception e = new Exception("Exceeded Netty Web Socket maximum buffer size of " + getMaxBufferSize());
+ onError(e);
+ this.close();
+ return;
+ }
+ }
 
- if (last) {
- WebSocketByteListener.class.cast(l).onMessage(byteBuffer.toByteArray());
- byteBuffer = null;
- textBuffer = null;
+ for (WebSocketListener l : listeners) {
+ if (l instanceof WebSocketByteListener) {
+ try {
+ if (!last) {
+ WebSocketByteListener.class.cast(l).onFragment(message, last);
+ } else {
+ if (byteBuffer.size() > 0) {
+ byteBuffer.write(message);
+ WebSocketByteListener.class.cast(l).onFragment(message, last);
+ WebSocketByteListener.class.cast(l).onMessage(byteBuffer.toByteArray());
+ } else {
+ WebSocketByteListener.class.cast(l).onMessage(message);
+ }
  }
  } catch (Exception ex) {
  l.onError(ex);
  }
  }
  }
+
+ if (last) {
+ byteBuffer.reset();
+ }
  }
 
  public void onTextFragment(String message, boolean last) {
- for (WebSocketListener l : listeners) {
- if (l instanceof WebSocketTextListener) {
- try {
- WebSocketTextListener.class.cast(l).onFragment(message, last);
 
- if (textBuffer == null) {
- textBuffer = new StringBuilder();
- }
-
- textBuffer.append(message);
+ if (!last) {
+ textBuffer.append(message);
 
- if (textBuffer.length() > maxBufferSize) {
- Exception e = new Exception("Exceeded Netty Web Socket maximum buffer size of " + getMaxBufferSize());
- l.onError(e);
- this.close();
- return;
- }
+ if (textBuffer.length() > maxBufferSize) {
+ textBuffer.setLength(0);
+ Exception e = new Exception("Exceeded Netty Web Socket maximum buffer size of " + getMaxBufferSize());
+ onError(e);
+ this.close();
+ return;
+ }
+ }
 
- if (last) {
- WebSocketTextListener.class.cast(l).onMessage(textBuffer.toString());
- byteBuffer = null;
- textBuffer = null;
+ for (WebSocketListener l : listeners) {
+ if (l instanceof WebSocketTextListener) {
+ try {
+ if (!last) {
+ WebSocketTextListener.class.cast(l).onFragment(message, last);
+ } else {
+ if (textBuffer.length() > 0) {
+ WebSocketTextListener.class.cast(l).onFragment(message, last);
+
+ WebSocketTextListener.class.cast(l).onMessage(textBuffer.append(message).toString());
+ } else {
+ WebSocketTextListener.class.cast(l).onMessage(message);
+ }
  }
  } catch (Exception ex) {
  l.onError(ex);
  }
  }
  }
+
+ if (last) {
+ textBuffer.setLength(0);
+ }
  }
 
  public void onError(Throwable t) {