multy node with redisson receive multy same message

README.md

@@ -1,3 +1,4 @@
+
 Netty-socketio Overview
 ===
 This project is an open-source Java implementation of [Socket.IO](http://socket.io/) server. Based on [Netty](http://netty.io/) server framework.  

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;
 
 /**
- * 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();
-    }
-
-    @Override
-    public void disconnect() {
-        for (SocketIOClient client : clients) {
-            client.disconnect();
-        }
-    }
-
-    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);
-    }
-
-    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();
-    }
-
-
+public interface BroadcastOperations extends ClientOperations {
+
+	public Collection<SocketIOClient> getClients();
+
+	public <T> void send(Packet packet, BroadcastAckCallback<T> ackCallback);
+
+	public void sendEvent(String name, SocketIOClient excludedClient, Object... data);
+
+	public <T> void sendEvent(String name, Object data, BroadcastAckCallback<T> ackCallback);
+
+	public <T> void sendEvent(String name, Object data, SocketIOClient excludedClient, BroadcastAckCallback<T> ackCallback);
+
 }

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

@@ -0,0 +1,118 @@
+/**
+ * Copyright 2012 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 java.util.Collection;
+import java.util.HashSet;
+import java.util.Set;
+
+import com.corundumstudio.socketio.protocol.Packet;
+
+/**
+ * Fully thread-safe.
+ *
+ */
+public class MultiRoomBroadcastOperations implements BroadcastOperations {
+
+	private Collection<BroadcastOperations> broadcastOperations;
+
+    public MultiRoomBroadcastOperations( Collection<BroadcastOperations> broadcastOperations ) {
+    	this.broadcastOperations = broadcastOperations;
+    }
+
+    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 void send(Packet packet) {
+    	if( this.broadcastOperations == null || this.broadcastOperations.size() == 0 ) {
+    		return;
+    	}
+    	for( BroadcastOperations b : this.broadcastOperations ) {
+    		b.send( packet );
+    	}
+    }
+
+    @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 disconnect() {
+    	if( this.broadcastOperations == null || this.broadcastOperations.size() == 0 ) {
+    		return;
+    	}
+    	for( BroadcastOperations b : this.broadcastOperations ) {
+    		b.disconnect();
+    	}
+    }
+
+    @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 void sendEvent(String name, Object... data) {
+    	if( this.broadcastOperations == null || this.broadcastOperations.size() == 0 ) {
+    		return;
+    	}
+    	for( BroadcastOperations b : this.broadcastOperations ) {
+    		b.sendEvent( name, 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 );
+    	}
+    }
+
+
+}