add kick remote interface

mpush-api/src/main/java/com/shinemo/mpush/api/event/RouterChangeEvent.java

@@ -0,0 +1,16 @@
+package com.shinemo.mpush.api.event;
+
+import com.shinemo.mpush.api.router.Router;
+
+/**
+ * Created by ohun on 2016/1/4.
+ */
+public class RouterChangeEvent implements Event {
+    public final String userId;
+    public final Router<?> router;
+
+    public RouterChangeEvent(String userId, Router<?> router) {
+        this.userId = userId;
+        this.router = router;
+    }
+}

mpush-client/src/main/java/com/shinemo/mpush/client/PushClient.java

@@ -3,14 +3,7 @@
 import com.shinemo.mpush.api.Client;
 import com.shinemo.mpush.api.PushSender;
 import com.shinemo.mpush.api.connection.Connection;
-import com.shinemo.mpush.api.router.ClientLocation;
-import com.shinemo.mpush.common.message.gateway.GatewayPushMessage;
-import com.shinemo.mpush.common.router.RemoteRouter;
-import com.shinemo.mpush.common.router.RemoteRouterManager;
-import com.shinemo.mpush.common.router.RouterCenter;
 import com.shinemo.mpush.netty.client.NettyClientFactory;
-import io.netty.channel.ChannelFuture;
-import io.netty.channel.ChannelFutureListener;
 
 import java.util.Collection;
 
@@ -28,7 +21,7 @@ public void init() throws Exception {
         this.clientFactory = NettyClientFactory.INSTANCE;
     }
 
-    private Connection getConnection(String ip) {
+    public Connection getConnection(String ip) {
         try {
             Client client = clientFactory.get(ip, port);
             if (client == null) {
@@ -64,29 +57,5 @@ public void send(String content, Collection<String> userIds, Callback callback)
     }
 
 
-    public void send(String content, final String userId, final PushRequest callback) {
-        RemoteRouterManager remoteRouterManager = RouterCenter.INSTANCE.getRemoteRouterManager();
-        RemoteRouter router = remoteRouterManager.lookup(userId);
-        if (router == null) {
-            callback.onOffline(userId);
-            return;
-        }
-        ClientLocation location = router.getRouteValue();
-        Connection connection = getConnection(location.getHost());
-        if (connection == null || !connection.isConnected()) {
-            callback.onFailure(userId);
-            return;
-        }
-        GatewayPushMessage pushMessage = new GatewayPushMessage(userId, content, connection);
-        PushRequestBus.INSTANCE.register(pushMessage.getSessionId(), callback);
-        pushMessage.send(new ChannelFutureListener() {
-            @Override
-            public void operationComplete(ChannelFuture future) throws Exception {
-                if (future.isSuccess()) {
-                } else {
-                    callback.onFailure(userId);
-                }
-            }
-        });
-    }
+
 }

mpush-client/src/main/java/com/shinemo/mpush/client/PushRequest.java

@@ -1,18 +1,30 @@
 package com.shinemo.mpush.client;
 
 import com.shinemo.mpush.api.PushSender;
+import com.shinemo.mpush.api.connection.Connection;
+import com.shinemo.mpush.api.router.ClientLocation;
+import com.shinemo.mpush.common.message.gateway.GatewayPushMessage;
+import com.shinemo.mpush.common.router.ConnectionRouterManager;
+import com.shinemo.mpush.common.router.RemoteRouter;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelFutureListener;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * Created by ohun on 2015/12/30.
  */
 public class PushRequest implements PushSender.Callback, Runnable {
+    private static final Logger LOGGER = LoggerFactory.getLogger(PushRequest.class);
     private PushSender.Callback callback;
     private String userId;
     private String content;
     private long timeout;
     private PushClient pushClient;
     private int status = 0;
     private long timeout_;
+    private int sessionId;
+    private long sendTime;
 
     public PushRequest(PushClient pushClient) {
         this.pushClient = pushClient;
@@ -42,6 +54,14 @@ public PushRequest setTimeout(long timeout) {
         return this;
     }
 
+    public void setSessionId(int sessionId) {
+        this.sessionId = sessionId;
+    }
+
+    public int getSessionId() {
+        return sessionId;
+    }
+
     @Override
     public void onSuccess(String userId) {
         submit(1);
@@ -64,10 +84,11 @@ public void onTimeout(String userId) {
 
     private void submit(int status) {
         this.status = status;
+        if (sessionId > 0) PushRequestBus.INSTANCE.remove(sessionId);
         if (callback != null) {
             PushRequestBus.INSTANCE.getExecutor().execute(this);
         } else {
-
+            LOGGER.warn("callback is null");
         }
     }
 
@@ -85,6 +106,10 @@ public void run() {
         }
     }
 
+    public boolean isTimeout() {
+        return System.currentTimeMillis() > timeout_;
+    }
+
     public void timeout() {
         onTimeout(userId);
     }
@@ -103,15 +128,40 @@ public void offline() {
 
     public void send() {
         this.timeout_ = timeout + System.currentTimeMillis();
-        pushClient.send(content, userId, this);
+        sendToConnectionServer();
     }
 
     public void redirect() {
         this.timeout_ = timeout + System.currentTimeMillis();
-        pushClient.send(content, userId, this);
+        ConnectionRouterManager.INSTANCE.invalidateLocalCache(userId);
+        sendToConnectionServer();
+        LOGGER.warn("user route has changed, userId={}, content={}", userId, content);
     }
 
-    public boolean isTimeout() {
-        return System.currentTimeMillis() > timeout_;
+    private void sendToConnectionServer() {
+        RemoteRouter router = ConnectionRouterManager.INSTANCE.lookup(userId);
+        if (router == null) {
+            this.onOffline(userId);
+            return;
+        }
+        ClientLocation location = router.getRouteValue();
+        Connection connection = pushClient.getConnection(location.getHost());
+        if (connection == null || !connection.isConnected()) {
+            this.onFailure(userId);
+            return;
+        }
+        GatewayPushMessage pushMessage = new GatewayPushMessage(userId, content, connection);
+        pushMessage.send(new ChannelFutureListener() {
+            @Override
+            public void operationComplete(ChannelFuture future) throws Exception {
+                if (future.isSuccess()) {
+                    sendTime = System.currentTimeMillis();
+                } else {
+                    PushRequest.this.onFailure(userId);
+                }
+            }
+        });
+        this.sessionId = pushMessage.getSessionId();
+        PushRequestBus.INSTANCE.add(this);
     }
 }

mpush-client/src/main/java/com/shinemo/mpush/client/PushRequestBus.java

@@ -16,8 +16,8 @@ public PushRequestBus() {
         scheduledExecutor.scheduleAtFixedRate(this, 1, 3, TimeUnit.SECONDS);
     }
 
-    public void register(int reqId, PushRequest callback) {
-        requests.put(reqId, callback);
+    public void add(PushRequest request) {
+        requests.put(request.getSessionId(), request);
     }
 
     public PushRequest remove(int reqId) {

mpush-common/src/main/java/com/shinemo/mpush/common/router/ConnectionRouterManager.java

@@ -0,0 +1,49 @@
+package com.shinemo.mpush.common.router;
+
+import com.google.common.cache.Cache;
+import com.google.common.cache.CacheBuilder;
+
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Created by ohun on 2016/1/4.
+ */
+public class ConnectionRouterManager extends RemoteRouterManager {
+    public static final ConnectionRouterManager INSTANCE = new ConnectionRouterManager();
+    // TODO: 2015/12/30 可以增加一层本地缓存，防止疯狂查询redis, 但是要注意失效问题及数据不一致问题
+    private final Cache<String, RemoteRouter> cache = CacheBuilder
+            .newBuilder()
+            .expireAfterWrite(5, TimeUnit.MINUTES)
+            .expireAfterAccess(5, TimeUnit.MINUTES)
+            .build();
+
+
+    @Override
+    public RemoteRouter register(String userId, RemoteRouter route) {
+        RemoteRouter old = cache.getIfPresent(userId);
+        cache.put(userId, route);
+        return old;
+    }
+
+    @Override
+    public boolean unRegister(String userId) {
+        cache.invalidate(userId);
+        return true;
+    }
+
+    @Override
+    public RemoteRouter lookup(String userId) {
+        return cache.getIfPresent(userId);
+    }
+
+    /**
+     * 如果推送失败，可能是缓存不一致了，可以让本地缓存失效
+     * <p>
+     * 失效对应的本地缓存
+     *
+     * @param userId
+     */
+    public void invalidateLocalCache(String userId) {
+        cache.invalidate(userId);
+    }
+}

mpush-common/src/main/java/com/shinemo/mpush/common/router/RemoteRouterManager.java

@@ -1,49 +1,24 @@
 package com.shinemo.mpush.common.router;
 
-import com.google.common.cache.Cache;
-import com.google.common.cache.CacheBuilder;
 import com.shinemo.mpush.api.router.RouterManager;
 
-import java.util.concurrent.TimeUnit;
-
 /**
  * Created by ohun on 2015/12/23.
  */
 public class RemoteRouterManager implements RouterManager<RemoteRouter> {
-    // TODO: 2015/12/30 可以增加一层本地缓存，防止疯狂查询redis, 但是要注意失效问题及数据不一致问题
-    private final Cache<String, RemoteRouter> cache = CacheBuilder
-            .newBuilder()
-            .expireAfterWrite(5, TimeUnit.MINUTES)
-            .expireAfterAccess(5, TimeUnit.MINUTES)
-            .build();
-
 
     @Override
     public RemoteRouter register(String userId, RemoteRouter route) {
-        RemoteRouter old = cache.getIfPresent(userId);
-        cache.put(userId, route);
-        return old;
+        return null;
     }
 
     @Override
     public boolean unRegister(String userId) {
-        cache.invalidate(userId);
         return true;
     }
 
     @Override
     public RemoteRouter lookup(String userId) {
-        return cache.getIfPresent(userId);
-    }
-
-    /**
-     * 如果推送失败，可能是缓存不一致了，可以让本地缓存失效
-     * <p>
-     * 失效对应的本地缓存
-     *
-     * @param userId
-     */
-    public void invalidateLocalCache(String userId) {
-        cache.invalidate(userId);
+        return null;
     }
 }

mpush-core/src/main/java/com/shinemo/mpush/core/handler/BindUserHandler.java

@@ -8,7 +8,7 @@
 import com.shinemo.mpush.common.message.OkMessage;
 import com.shinemo.mpush.api.protocol.Packet;
 import com.shinemo.mpush.common.handler.BaseMessageHandler;
-import com.shinemo.mpush.common.router.RouterCenter;
+import com.shinemo.mpush.core.router.RouterCenter;
 
 /**
  * Created by ohun on 2015/12/23.

mpush-core/src/main/java/com/shinemo/mpush/core/handler/GatewayPushHandler.java

@@ -9,7 +9,7 @@
 import com.shinemo.mpush.api.protocol.Packet;
 import com.shinemo.mpush.api.router.Router;
 import com.shinemo.mpush.common.handler.BaseMessageHandler;
-import com.shinemo.mpush.common.router.RouterCenter;
+import com.shinemo.mpush.core.router.RouterCenter;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelFutureListener;
 

mpush-core/src/main/java/com/shinemo/mpush/core/router/KickRemoteMsg.java

@@ -0,0 +1,19 @@
+package com.shinemo.mpush.core.router;
+
+/**
+ * Created by ohun on 2016/1/4.
+ */
+public class KickRemoteMsg {
+    public String userId;
+    public String deviceId;
+    public String srcServer;
+
+    @Override
+    public String toString() {
+        return "KickRemoteMsg{" +
+                "userId='" + userId + '\'' +
+                ", deviceId='" + deviceId + '\'' +
+                ", srcServer='" + srcServer + '\'' +
+                '}';
+    }
+}

mpush-common/src/main/java/com/shinemo/mpush/common/router/LocalRouter.java → mpush-core/src/main/java/com/shinemo/mpush/core/router/LocalRouter.java

@@ -1,4 +1,4 @@
-package com.shinemo.mpush.common.router;
+package com.shinemo.mpush.core.router;
 
 import com.shinemo.mpush.api.connection.Connection;
 import com.shinemo.mpush.api.router.Router;

mpush-common/src/main/java/com/shinemo/mpush/common/router/LocalRouterManager.java → mpush-core/src/main/java/com/shinemo/mpush/core/router/LocalRouterManager.java

@@ -1,4 +1,4 @@
-package com.shinemo.mpush.common.router;
+package com.shinemo.mpush.core.router;
 
 import com.shinemo.mpush.api.router.RouterManager;