Branch netty socketio 1.7.12 update MemoryPubSubStore code

pom.xml

@@ -194,6 +194,12 @@
         <version>3.4.3</version>
         <scope>provided</scope>
     </dependency>
+    <dependency>
+		<groupId>com.google.guava</groupId>
+		<artifactId>guava</artifactId>
+		<version>18.0</version>
+		<scope>provided</scope>
+	</dependency>
  </dependencies>
 
 

src/main/java/com/corundumstudio/socketio/store/MemoryPubSubStore.java

@@ -15,23 +15,46 @@
  */
 package com.corundumstudio.socketio.store;
 
+import java.util.Map;
+import java.util.concurrent.Executors;
+
 import com.corundumstudio.socketio.store.pubsub.PubSubListener;
 import com.corundumstudio.socketio.store.pubsub.PubSubMessage;
 import com.corundumstudio.socketio.store.pubsub.PubSubStore;
 import com.corundumstudio.socketio.store.pubsub.PubSubType;
 
-public class MemoryPubSubStore implements PubSubStore {
+import com.google.common.collect.Maps;
+import com.google.common.eventbus.AsyncEventBus;
 
+public class MemoryPubSubStore implements PubSubStore {
+
+	private final AsyncEventBus asyncEventBus = new AsyncEventBus(Executors.newFixedThreadPool(3));
+
+    private final Map<PubSubType,PubSubListener<?>> subtypeListener = Maps.newHashMap();
+
+    private final Long nodeId;
+
+    public MemoryPubSubStore(Long nodeId){
+        this.nodeId = nodeId;
+    }
+
     @Override
     public void publish(PubSubType type, PubSubMessage msg) {
+		msg.setNodeId(nodeId);
+        asyncEventBus.post(msg);
     }
 
     @Override
     public <T extends PubSubMessage> void subscribe(PubSubType type, PubSubListener<T> listener, Class<T> clazz) {
+		subtypeListener.put(type, listener);
+        asyncEventBus.register(listener);
     }
 
     @Override
     public void unsubscribe(PubSubType type) {
+		PubSubListener<?> ls = subtypeListener.get(type);
+        asyncEventBus.unregister(ls);
+        subtypeListener.remove(type);
     }
 
     @Override

src/main/java/com/corundumstudio/socketio/store/MemoryStoreFactory.java

@@ -25,7 +25,7 @@
 
 public class MemoryStoreFactory extends BaseStoreFactory {
 
-    private final MemoryPubSubStore pubSubMemoryStore = new MemoryPubSubStore();
+    private final MemoryPubSubStore pubSubMemoryStore = new MemoryPubSubStore(getNodeId());
 
     @Override
     public Store createStore(UUID sessionId) {

src/main/java/com/corundumstudio/socketio/store/pubsub/PubSubListener.java

@@ -15,9 +15,11 @@
  */
 package com.corundumstudio.socketio.store.pubsub;
 
+import com.google.common.eventbus.Subscribe;
 
 public interface PubSubListener<T> {
 
+    @Subscribe
     void onMessage(T data);
 
 }