Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

multy node with redisson receive multy same message #552

Open
wants to merge 8 commits into
base: master
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions README.md
Original file line number Diff line number Diff line change
@@ -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.
Expand Down
125 changes: 13 additions & 112 deletions src/main/java/com/corundumstudio/socketio/BroadcastOperations.java
Original file line number Diff line number Diff line change
Expand Up @@ -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);

}
Original file line number Diff line number Diff line change
@@ -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 );
}
}


}
Loading