add redis subscribe

mpusher · Jan 5, 2016 · 1430718 · 1430718
1 parent 60952fe
commit 1430718
Show file tree

Hide file tree

Showing 6 changed files with 882 additions and 830 deletions.
diff --git a/mpush-core/src/main/java/com/shinemo/mpush/core/router/RouterChangeListener.java b/mpush-core/src/main/java/com/shinemo/mpush/core/router/RouterChangeListener.java
@@ -9,6 +9,10 @@
 import com.shinemo.mpush.common.EventBus;
 import com.shinemo.mpush.common.message.KickUserMessage;
 import com.shinemo.mpush.common.router.RemoteRouter;
+import com.shinemo.mpush.tools.Jsons;
+import com.shinemo.mpush.tools.redis.listener.MessageListener;
+import com.shinemo.mpush.tools.redis.manage.RedisManage;
+import com.shinemo.mpush.tools.redis.pubsub.Subscriber;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelFutureListener;
 import org.slf4j.Logger;
@@ -17,12 +21,13 @@
 /**
  * Created by ohun on 2016/1/4.
  */
-public class RouterChangeListener {
+public class RouterChangeListener implements MessageListener {
     private static final Logger LOGGER = LoggerFactory.getLogger(RouterChangeListener.class);
+    public static final String KICK_CHANNEL = "__kick__";
 
     public RouterChangeListener() {
         EventBus.INSTANCE.register(this);
-        // TODO: 2016/1/4 register this to redis server
+        RedisManage.subscribe(this, KICK_CHANNEL);
     }
 
     @Subscribe
@@ -61,10 +66,9 @@ public void kickRemote(String userId, RemoteRouter router) {
         msg.deviceId = location.getDeviceId();
         msg.srcServer = location.getHost();
         msg.userId = userId;
-        // TODO: 2016/1/4 publish kick remote user msg to redis
+        RedisManage.publish(KICK_CHANNEL, msg);
     }
 
-    // TODO: 2016/1/4  receive msg from redis
     public void onReceiveKickRemoteMsg(KickRemoteMsg msg) {
         String userId = msg.userId;
         LocalRouterManager routerManager = RouterCenter.INSTANCE.getLocalRouterManager();
@@ -77,4 +81,18 @@ public void onReceiveKickRemoteMsg(KickRemoteMsg msg) {
             LOGGER.warn("no local router find, kick failure, msg={}", msg);
         }
     }
+
+    @Override
+    public void onMessage(String channel, String message) {
+        if (KICK_CHANNEL.equals(channel)) {
+            KickRemoteMsg msg = Jsons.fromJson(message, KickRemoteMsg.class);
+            if (msg != null) {
+                onReceiveKickRemoteMsg(msg);
+            } else {
+                LOGGER.warn("receive an error kick message={}", message);
+            }
+        } else {
+            LOGGER.warn("receive an error redis channel={}", channel);
+        }
+    }
 }