Merge pull request #771 from GaryLeung922/hotfix

Hotfix：multy node with redisson receive multy same message

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

@@ -15,123 +15,24 @@
  */
 package com.corundumstudio.socketio;
 
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.Set;
-
-import com.corundumstudio.socketio.misc.IterableCollection;
-import com.corundumstudio.socketio.namespace.Namespace;
 import com.corundumstudio.socketio.protocol.Packet;
-import com.corundumstudio.socketio.protocol.PacketType;
-import com.corundumstudio.socketio.store.StoreFactory;
-import com.corundumstudio.socketio.store.pubsub.DispatchMessage;
-import com.corundumstudio.socketio.store.pubsub.PubSubType;
+
+import java.util.Collection;
 
 /**
- * Fully thread-safe.
+ * broadcast interface
  *
  */
-public class BroadcastOperations implements ClientOperations {
-
-    private final Iterable<SocketIOClient> clients;
-    private final StoreFactory storeFactory;
-
-    public BroadcastOperations(Iterable<SocketIOClient> clients, StoreFactory storeFactory) {
-        super();
-        this.clients = clients;
-        this.storeFactory = storeFactory;
-    }
-
-    private void dispatch(Packet packet) {
-        Map<String, Set<String>> namespaceRooms = new HashMap<String, Set<String>>();
-        for (SocketIOClient socketIOClient : clients) {
-            Namespace namespace = (Namespace)socketIOClient.getNamespace();
-            Set<String> rooms = namespace.getRooms(socketIOClient);
-            
-            Set<String> roomsList = namespaceRooms.get(namespace.getName());
-            if (roomsList == null) {
-                roomsList = new HashSet<String>();
-                namespaceRooms.put(namespace.getName(), roomsList);
-            }
-            roomsList.addAll(rooms);
-        }
-        for (Entry<String, Set<String>> entry : namespaceRooms.entrySet()) {
-            for (String room : entry.getValue()) {
-                storeFactory.pubSubStore().publish(PubSubType.DISPATCH, new DispatchMessage(room, packet, entry.getKey()));
-            }
-        }
-    }
-
-    public Collection<SocketIOClient> getClients() {
-        return new IterableCollection<SocketIOClient>(clients);
-    }
-
-    @Override
-    public void send(Packet packet) {
-        for (SocketIOClient client : clients) {
-            client.send(packet);
-        }
-        dispatch(packet);
-    }
-
-    public <T> void send(Packet packet, BroadcastAckCallback<T> ackCallback) {
-        for (SocketIOClient client : clients) {
-            client.send(packet, ackCallback.createClientCallback(client));
-        }
-        ackCallback.loopFinished();
-    }
+public interface BroadcastOperations extends ClientOperations {
 
-    @Override
-    public void disconnect() {
-        for (SocketIOClient client : clients) {
-            client.disconnect();
-        }
-    }
+    Collection<SocketIOClient> getClients();
 
-    public void sendEvent(String name, SocketIOClient excludedClient, Object... data) {
-        Packet packet = new Packet(PacketType.MESSAGE);
-        packet.setSubType(PacketType.EVENT);
-        packet.setName(name);
-        packet.setData(Arrays.asList(data));
+    <T> void send(Packet packet, BroadcastAckCallback<T> ackCallback);
 
-        for (SocketIOClient client : clients) {
-            if (client.getSessionId().equals(excludedClient.getSessionId())) {
-                continue;
-            }
-            client.send(packet);
-        }
-        dispatch(packet);
-    }
-    
-    @Override
-    public void sendEvent(String name, Object... data) {
-        Packet packet = new Packet(PacketType.MESSAGE);
-        packet.setSubType(PacketType.EVENT);
-        packet.setName(name);
-        packet.setData(Arrays.asList(data));
-        send(packet);
-    }
+    void sendEvent(String name, SocketIOClient excludedClient, Object... data);
 
-    public <T> void sendEvent(String name, Object data, BroadcastAckCallback<T> ackCallback) {
-        for (SocketIOClient client : clients) {
-            client.sendEvent(name, ackCallback.createClientCallback(client), data);
-        }
-        ackCallback.loopFinished();
-    }
-    
-    public <T> void sendEvent(String name, Object data, SocketIOClient excludedClient, BroadcastAckCallback<T> ackCallback) {
-        for (SocketIOClient client : clients) {
-            if (client.getSessionId().equals(excludedClient.getSessionId())) {
-                continue;
-            }
-            client.sendEvent(name, ackCallback.createClientCallback(client), data);
-        }
-        ackCallback.loopFinished();
-    }
+    <T> void sendEvent(String name, Object data, BroadcastAckCallback<T> ackCallback);
 
+    <T> void sendEvent(String name, Object data, SocketIOClient excludedClient, BroadcastAckCallback<T> ackCallback);
 
 }

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

@@ -0,0 +1,117 @@
+/**
+ * Copyright (c) 2012-2019 Nikita Koksharov
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.corundumstudio.socketio;
+
+import com.corundumstudio.socketio.protocol.Packet;
+
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.Set;
+
+/**
+ * @Author: liangjiaqi
+ * @Date: 2020/8/8 6:02 PM
+ */
+public class MultiRoomBroadcastOperations implements BroadcastOperations {
+
+    private Collection<BroadcastOperations> broadcastOperations;
+
+    public MultiRoomBroadcastOperations(Collection<BroadcastOperations> broadcastOperations) {
+        this.broadcastOperations = broadcastOperations;
+    }
+
+    @Override
+    public Collection<SocketIOClient> getClients() {
+        Set<SocketIOClient> clients = new HashSet<SocketIOClient>();
+        if( this.broadcastOperations == null || this.broadcastOperations.size() == 0 ) {
+            return clients;
+        }
+        for( BroadcastOperations b : this.broadcastOperations ) {
+            clients.addAll( b.getClients() );
+        }
+        return clients;
+    }
+
+    @Override
+    public <T> void send(Packet packet, BroadcastAckCallback<T> ackCallback) {
+        if( this.broadcastOperations == null || this.broadcastOperations.size() == 0 ) {
+            return;
+        }
+        for( BroadcastOperations b : this.broadcastOperations ) {
+            b.send( packet, ackCallback );
+        }
+    }
+
+    @Override
+    public void sendEvent(String name, SocketIOClient excludedClient, Object... data) {
+        if( this.broadcastOperations == null || this.broadcastOperations.size() == 0 ) {
+            return;
+        }
+        for( BroadcastOperations b : this.broadcastOperations ) {
+            b.sendEvent( name, excludedClient, data );
+        }
+    }
+
+    @Override
+    public <T> void sendEvent(String name, Object data, BroadcastAckCallback<T> ackCallback) {
+        if( this.broadcastOperations == null || this.broadcastOperations.size() == 0 ) {
+            return;
+        }
+        for( BroadcastOperations b : this.broadcastOperations ) {
+            b.sendEvent( name, data, ackCallback );
+        }
+    }
+
+    @Override
+    public <T> void sendEvent(String name, Object data, SocketIOClient excludedClient, BroadcastAckCallback<T> ackCallback) {
+        if( this.broadcastOperations == null || this.broadcastOperations.size() == 0 ) {
+            return;
+        }
+        for( BroadcastOperations b : this.broadcastOperations ) {
+            b.sendEvent( name, data, excludedClient, ackCallback );
+        }
+    }
+
+    @Override
+    public void send(Packet packet) {
+        if( this.broadcastOperations == null || this.broadcastOperations.size() == 0 ) {
+            return;
+        }
+        for( BroadcastOperations b : this.broadcastOperations ) {
+            b.send( packet );
+        }
+    }
+
+    @Override
+    public void disconnect() {
+        if( this.broadcastOperations == null || this.broadcastOperations.size() == 0 ) {
+            return;
+        }
+        for( BroadcastOperations b : this.broadcastOperations ) {
+            b.disconnect();
+        }
+    }
+
+    @Override
+    public void sendEvent(String name, Object... data) {
+        if( this.broadcastOperations == null || this.broadcastOperations.size() == 0 ) {
+            return;
+        }
+        for( BroadcastOperations b : this.broadcastOperations ) {
+            b.sendEvent( name, data );
+        }
+    }
+}

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

@@ -0,0 +1,123 @@
+/**
+ * Copyright (c) 2012-2019 Nikita Koksharov
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.corundumstudio.socketio;
+
+import com.corundumstudio.socketio.misc.IterableCollection;
+import com.corundumstudio.socketio.protocol.Packet;
+import com.corundumstudio.socketio.protocol.PacketType;
+import com.corundumstudio.socketio.store.StoreFactory;
+import com.corundumstudio.socketio.store.pubsub.DispatchMessage;
+import com.corundumstudio.socketio.store.pubsub.PubSubType;
+
+import java.util.Arrays;
+import java.util.Collection;
+
+/**
+ * @Author: liangjiaqi
+ * @Date: 2020/8/8 6:08 PM
+ */
+public class SingleRoomBroadcastOperations implements BroadcastOperations {
+    private final String namespace;
+    private final String room;
+    private final Iterable<SocketIOClient> clients;
+    private final StoreFactory storeFactory;
+
+    public SingleRoomBroadcastOperations(String namespace, String room, Iterable<SocketIOClient> clients, StoreFactory storeFactory) {
+        super();
+        this.namespace = namespace;
+        this.room = room;
+        this.clients = clients;
+        this.storeFactory = storeFactory;
+    }
+
+    private void dispatch(Packet packet) {
+        this.storeFactory.pubSubStore().publish(
+                PubSubType.DISPATCH,
+                new DispatchMessage(this.room, packet, this.namespace));
+    }
+
+    @Override
+    public Collection<SocketIOClient> getClients() {
+        return new IterableCollection<SocketIOClient>(clients);
+    }
+
+    @Override
+    public void send(Packet packet) {
+        for (SocketIOClient client : clients) {
+            client.send(packet);
+        }
+        dispatch(packet);
+    }
+
+    @Override
+    public <T> void send(Packet packet, BroadcastAckCallback<T> ackCallback) {
+        for (SocketIOClient client : clients) {
+            client.send(packet, ackCallback.createClientCallback(client));
+        }
+        ackCallback.loopFinished();
+    }
+
+    @Override
+    public void disconnect() {
+        for (SocketIOClient client : clients) {
+            client.disconnect();
+        }
+    }
+
+    @Override
+    public void sendEvent(String name, SocketIOClient excludedClient, Object... data) {
+        Packet packet = new Packet(PacketType.MESSAGE);
+        packet.setSubType(PacketType.EVENT);
+        packet.setName(name);
+        packet.setData(Arrays.asList(data));
+
+        for (SocketIOClient client : clients) {
+            if (client.getSessionId().equals(excludedClient.getSessionId())) {
+                continue;
+            }
+            client.send(packet);
+        }
+        dispatch(packet);
+    }
+
+    @Override
+    public void sendEvent(String name, Object... data) {
+        Packet packet = new Packet(PacketType.MESSAGE);
+        packet.setSubType(PacketType.EVENT);
+        packet.setName(name);
+        packet.setData(Arrays.asList(data));
+        send(packet);
+    }
+
+    @Override
+    public <T> void sendEvent(String name, Object data, BroadcastAckCallback<T> ackCallback) {
+        for (SocketIOClient client : clients) {
+            client.sendEvent(name, ackCallback.createClientCallback(client), data);
+        }
+        ackCallback.loopFinished();
+    }
+
+    @Override
+    public <T> void sendEvent(String name, Object data, SocketIOClient excludedClient, BroadcastAckCallback<T> ackCallback) {
+        for (SocketIOClient client : clients) {
+            if (client.getSessionId().equals(excludedClient.getSessionId())) {
+                continue;
+            }
+            client.sendEvent(name, ackCallback.createClientCallback(client), data);
+        }
+        ackCallback.loopFinished();
+    }
+}

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

@@ -29,7 +29,9 @@ import io.netty.util.concurrent.Future;
 import io.netty.util.concurrent.FutureListener;
 
 import java.net.InetSocketAddress;
+import java.util.ArrayList;
 import java.util.Collection;
+import java.util.List;
 import java.util.UUID;
 
 import org.slf4j.Logger;
@@ -97,7 +99,16 @@ public class SocketIOServer implements ClientListeners {
     }
 
     public BroadcastOperations getBroadcastOperations() {
-        return new BroadcastOperations(getAllClients(), configCopy.getStoreFactory());
+        Collection<SocketIONamespace> namespaces = namespacesHub.getAllNamespaces();
+        List<BroadcastOperations> list = new ArrayList<BroadcastOperations>();
+        BroadcastOperations broadcast = null;
+        if( namespaces != null && namespaces.size() > 0 ) {
+            for( SocketIONamespace n : namespaces ) {
+                broadcast = n.getBroadcastOperations();
+                list.add( broadcast );
+            }
+        }
+        return new MultiRoomBroadcastOperations( list );
     }
 
     /**
@@ -108,8 +119,16 @@ public class SocketIOServer implements ClientListeners {
      * @return broadcast operations
      */
     public BroadcastOperations getRoomOperations(String room) {
-        Iterable<SocketIOClient> clients = namespacesHub.getRoomClients(room);
-        return new BroadcastOperations(clients, configCopy.getStoreFactory());
+        Collection<SocketIONamespace> namespaces = namespacesHub.getAllNamespaces();
+        List<BroadcastOperations> list = new ArrayList<BroadcastOperations>();
+        BroadcastOperations broadcast = null;
+        if( namespaces != null && namespaces.size() > 0 ) {
+            for( SocketIONamespace n : namespaces ) {
+                broadcast = n.getRoomOperations( room );
+                list.add( broadcast );
+            }
+        }
+        return new MultiRoomBroadcastOperations( list );
     }
 
     /**

src/main/java/com/corundumstudio/socketio/namespace/Namespace.java

@@ -31,6 +31,7 @@ import com.corundumstudio.socketio.AckRequest;
 import com.corundumstudio.socketio.BroadcastOperations;
 import com.corundumstudio.socketio.Configuration;
 import com.corundumstudio.socketio.MultiTypeArgs;
+import com.corundumstudio.socketio.SingleRoomBroadcastOperations;
 import com.corundumstudio.socketio.SocketIOClient;
 import com.corundumstudio.socketio.SocketIONamespace;
 import com.corundumstudio.socketio.annotation.ScannerEngine;
@@ -187,11 +188,10 @@ public class Namespace implements SocketIONamespace {
         Set<String> joinedRooms = client.getAllRooms();        
         allClients.remove(client.getSessionId());
 
-        leave(getName(), client.getSessionId());
-        storeFactory.pubSubStore().publish(PubSubType.LEAVE, new JoinLeaveMessage(client.getSessionId(), getName(), getName()));
-
+        // client must leave all rooms and publish the leave msg one by one on disconnect.
         for (String joinedRoom : joinedRooms) {
             leave(roomClients, joinedRoom, client.getSessionId());
+            storeFactory.pubSubStore().publish(PubSubType.LEAVE, new JoinLeaveMessage(client.getSessionId(), joinedRoom, getName()));
         }
         clientRooms.remove(client.getSessionId());
 
@@ -239,12 +239,12 @@ public class Namespace implements SocketIONamespace {
 
     @Override
     public BroadcastOperations getBroadcastOperations() {
-        return new BroadcastOperations(allClients.values(), storeFactory);
+        return new SingleRoomBroadcastOperations(getName(), getName(), allClients.values(), storeFactory);
     }
 
     @Override
     public BroadcastOperations getRoomOperations(String room) {
-        return new BroadcastOperations(getRoomClients(room), storeFactory);
+        return new SingleRoomBroadcastOperations(getName(), room, getRoomClients(room), storeFactory);
     }
 
     @Override