Reduce the impact

Author: procothesky

src/main/java/com/corundumstudio/socketio/BroadcastOperations.java

@@ -21,12 +21,23 @@
 
 /**
  * broadcast interface
- *
  */
 public interface BroadcastOperations extends ClientOperations {
 
     Collection<SocketIOClient> getClients();
 
+    /**
+     * {@link Packet#attachments} needs to be filled when sending byte[].
+     * Using {@link io.netty.buffer.Unpooled#wrappedBuffer(byte[])} to
+     * fill byte[] into {@link Packet#attachments} is the recommended way.
+     * Before using {@link Packet#addAttachment(io.netty.buffer.ByteBuf)},
+     * be sure to initialize the number of attachments with
+     * {@link Packet#initAttachments(int)})}
+     *
+     * @param packet
+     * @param ackCallback
+     * @param <T>
+     */
     <T> void send(Packet packet, BroadcastAckCallback<T> ackCallback);
 
     void sendEvent(String name, SocketIOClient excludedClient, Object... data);

src/main/java/com/corundumstudio/socketio/ClientOperations.java

@@ -27,6 +27,7 @@ public interface ClientOperations {
      * Send custom packet.
      * But {@link ClientOperations#sendEvent} method
      * usage is enough for most cases.
+     * If the Packet is sent by BroadcastOperations,
      * {@link Packet#attachments} needs to be filled when sending byte[].
      * Using {@link io.netty.buffer.Unpooled#wrappedBuffer(byte[])} to
      * fill byte[] into {@link Packet#attachments} is the recommended way.

src/main/java/com/corundumstudio/socketio/SingleRoomBroadcastOperations.java

@@ -85,12 +85,15 @@ public void disconnect() {
     }
 
     @Override
-    public void sendEvent(String name, SocketIOClient excludedClient, Object... data) {
+    public void sendEvent(String name, SocketIOClient excludedClient, @NonNull Object... data) {
         Packet packet = new Packet(PacketType.MESSAGE, EngineIOVersion.UNKNOWN);
         packet.setSubType(PacketType.EVENT);
         packet.setName(name);
         packet.setData(Arrays.asList(data));
 
+        // handle byte[] data
+        handleBytes(packet, data);
+
         for (SocketIOClient client : clients) {
             packet.setEngineIOVersion(client.getEngineIOVersion());
             if (client.getSessionId().equals(excludedClient.getSessionId())) {
@@ -109,6 +112,12 @@ public void sendEvent(String name, @NonNull Object... data) {
         packet.setData(Arrays.asList(data));
 
         // handle byte[] data
+        handleBytes(packet, data);
+
+        send(packet);
+    }
+
+    private static void handleBytes(Packet packet, Object[] data) {
         List<byte[]> bytes = Arrays.stream(data)
                 .filter(o -> o instanceof byte[])
                 .map(b -> (byte[]) b)
@@ -119,8 +128,6 @@ public void sendEvent(String name, @NonNull Object... data) {
             packet.initAttachments(bytes.size());
             bytes.stream().peek(b -> packet.addAttachment(Unpooled.wrappedBuffer(b)));
         }
-
-        send(packet);
     }
 
     @Override

src/main/java/com/corundumstudio/socketio/protocol/PacketEncoder.java

@@ -15,7 +15,7 @@
  */
 package com.corundumstudio.socketio.protocol;
 
-import com.corundumstudio.socketio.handler.ClientHead;
+import com.corundumstudio.socketio.Configuration;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.buffer.ByteBufOutputStream;
@@ -29,8 +29,6 @@
 import java.util.List;
 import java.util.Queue;
 
-import com.corundumstudio.socketio.Configuration;
-
 public class PacketEncoder {
 
     private static final byte[] BINARY_HEADER = "b4".getBytes(CharsetUtil.UTF_8);
@@ -254,18 +252,20 @@ public void encodePacket(Packet packet, ByteBuf buffer, ByteBufAllocator allocat
 
                     ByteBuf encBuf = null;
 
-                    if (packet.getSubType() == PacketType.ERROR) {
+                    PacketType subType = packet.getSubType();
+                    if (subType == PacketType.ERROR) {
                         encBuf = allocateBuffer(allocator);
 
                         ByteBufOutputStream out = new ByteBufOutputStream(encBuf);
                         jsonSupport.writeValue(out, packet.getData());
                     }
 
-                    if (packet.getSubType() == PacketType.EVENT
-                            || packet.getSubType() == PacketType.ACK) {
+                    PacketType tmpSubType = subType;
+                    if (subType == PacketType.EVENT
+                            || subType == PacketType.ACK) {
 
                         List<Object> values = new ArrayList<Object>();
-                        if (packet.getSubType() == PacketType.EVENT) {
+                        if (subType == PacketType.EVENT) {
                             values.add(packet.getName());
                         }
 
@@ -277,21 +277,30 @@ public void encodePacket(Packet packet, ByteBuf buffer, ByteBufAllocator allocat
                         jsonSupport.writeValue(out, values);
 
                         if (!jsonSupport.getArrays().isEmpty()) {
-                            packet.setSubType(packet.getSubType() == PacketType.ACK
+                            // If the Packet is sent by BroadcastOperations,
+                            // there is a problem of concurrent initialization for the same Packet.
+                            // Please initAttachment when creating the Packet to avoid this problem.
+                            if (!packet.hasAttachments()) {
+                                packet.initAttachments(jsonSupport.getArrays().size());
+                                for (byte[] array : jsonSupport.getArrays()) {
+                                    packet.addAttachment(Unpooled.wrappedBuffer(array));
+                                }
+                            }
+                            tmpSubType = (subType == PacketType.ACK
                                     ? PacketType.BINARY_ACK : PacketType.BINARY_EVENT);
                         }
                     }
 
-                    byte subType = toChar(packet.getSubType().getValue());
-                    buf.writeByte(subType);
+                    byte subTypeByte = toChar(tmpSubType.getValue());
+                    buf.writeByte(subTypeByte);
 
                     if (packet.hasAttachments()) {
                         byte[] ackId = toChars(packet.getAttachments().size());
                         buf.writeBytes(ackId);
                         buf.writeByte('-');
                     }
 
-                    if (packet.getSubType() == PacketType.CONNECT) {
+                    if (subType == PacketType.CONNECT) {
                         if (!packet.getNsp().isEmpty()) {
                             buf.writeBytes(packet.getNsp().getBytes(CharsetUtil.UTF_8));
                         }