add Admin command

mpusher · May 24, 2016 · 717dba1 · 717dba1
1 parent 84a5181
commit 717dba1
Show file tree

Hide file tree

Showing 24 changed files with 201 additions and 382 deletions.
diff --git a/conf/conf-dev.properties b/conf/conf-dev.properties
@@ -0,0 +1 @@
+log.level=debug
diff --git a/conf/conf-pub.properties b/conf/conf-pub.properties
@@ -0,0 +1 @@
+log.level=warn
diff --git a/mpush-boot/assembly.xml b/mpush-boot/assembly.xml
@@ -1,7 +1,7 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <assembly>
     <id>release-${mpush.version}</id>
-    <baseDirectory>mpush</baseDirectory>
+    <baseDirectory>mpush-${mpush.version}</baseDirectory>
     <includeBaseDirectory>true</includeBaseDirectory>
     <formats>
         <format>tar.gz</format>

diff --git a/mpush-boot/src/main/java/com/mpush/bootstrap/Main.java b/mpush-boot/src/main/java/com/mpush/bootstrap/Main.java
@@ -25,17 +25,17 @@ public class Main {
 
     public static void main(String[] args) {
         Logs.init();
-        Logs.Console.info("launch app...");
+        Logs.Console.error("launch app...");
         ServerLauncher launcher = new ServerLauncher();
         launcher.start();
         addHook(launcher);
     }
 
-    private static void addHook(final ServerLauncher serverBoot) {
+    private static void addHook(final ServerLauncher launcher) {
         Runtime.getRuntime().addShutdownHook(new Thread() {
             public void run() {
-                serverBoot.stop();
+                launcher.stop();
-                Logs.Console.info("jvm exit all server stopped...");
+                Logs.Console.error("jvm exit, all server stopped...");
             }
         });
     }

diff --git a/mpush-boot/src/main/java/com/mpush/bootstrap/ServerLauncher.java b/mpush-boot/src/main/java/com/mpush/bootstrap/ServerLauncher.java
@@ -38,11 +38,11 @@ public class ServerLauncher {
 
     private final ZKServerNode gsNode = ZKServerNode.gsNode();
 
-    private final Server connectServer = new ConnectionServer(csNode.getPort());
+    private final ConnectionServer connectServer = new ConnectionServer(csNode.getPort());
 
-    private final Server gatewayServer = new GatewayServer(gsNode.getPort());
+    private final GatewayServer gatewayServer = new GatewayServer(gsNode.getPort());
 
-    private final Server adminServer = new AdminServer(CC.mp.net.admin_server_port);
+    private final AdminServer adminServer = new AdminServer(CC.mp.net.admin_server_port, connectServer, gatewayServer);
 
 
     public void start() {
@@ -60,7 +60,7 @@ public void start() {
     }
 
     public void stop() {
-        stopServer(gatewayServer);
+        stopServer(connectServer);
         stopServer(gatewayServer);
         stopServer(adminServer);
     }

diff --git a/mpush-boot/src/main/java/com/mpush/bootstrap/job/BootChain.java b/mpush-boot/src/main/java/com/mpush/bootstrap/job/BootChain.java
@@ -41,7 +41,7 @@ private BootJob first() {
         return new BootJob() {
             @Override
             public void run() {
-                Logs.Console.info("begin run bootstrap chain...");
+                Logs.Console.error("begin run bootstrap chain...");
                 next();
             }
         };

diff --git a/mpush-boot/src/main/java/com/mpush/bootstrap/job/BootJob.java b/mpush-boot/src/main/java/com/mpush/bootstrap/job/BootJob.java
@@ -33,7 +33,7 @@ public abstract class BootJob {
 
     public void next() {
         if (next != null) {
-            Logs.Console.info("run next bootstrap job [" + next.getClass().getSimpleName() + "]");
+            Logs.Console.error("run next bootstrap job [" + next.getClass().getSimpleName() + "]");
             next.run();
         }
     }

diff --git a/mpush-boot/src/main/java/com/mpush/bootstrap/job/LastBoot.java b/mpush-boot/src/main/java/com/mpush/bootstrap/job/LastBoot.java
@@ -31,7 +31,7 @@ public class LastBoot extends BootJob {
     @Override
     public void run() {
         UserManager.INSTANCE.clearUserOnlineData();
-        Logs.Console.info("end run bootstrap chain...");
+        Logs.Console.error("end run bootstrap chain...");
-        Logs.Console.info("app start success...");
+        Logs.Console.error("app start success...");
     }
 }
diff --git a/mpush-boot/src/main/java/com/mpush/bootstrap/job/ServerBoot.java b/mpush-boot/src/main/java/com/mpush/bootstrap/job/ServerBoot.java
@@ -50,7 +50,7 @@ public void run() {
                 server.start(new Server.Listener() {
                     @Override
                     public void onSuccess(Object... args) {
-                        Logs.Console.info("start " + serverName + " success listen:" + args[0]);
+                        Logs.Console.error("start " + serverName + " success listen:" + args[0]);
                         if (node != null) {
                             registerServerToZk(node.getZkPath(), Jsons.toJson(node));
                         }
@@ -70,6 +70,6 @@ public void onFailure(Throwable cause) {
     //step7  注册应用到zk
     public void registerServerToZk(String path, String value) {
         ZKClient.I.registerEphemeralSequential(path, value);
-        Logs.Console.info("register server node=" + value + " to zk name=" + path);
+        Logs.Console.error("register server node=" + value + " to zk name=" + path);
     }
 }
diff --git a/mpush-boot/src/main/resources/mpush.conf b/mpush-boot/src/main/resources/mpush.conf
@@ -1,2 +1,4 @@
-mp.log.level=debug
+mp.log.level=${log.level}
-mp.zk.namespace=mpush
+mp.security.private-key="MIIBNgIBADANBgkqhkiG9w0BAQEFAASCASAwggEcAgEAAoGBAKCE8JYKhsbydMPbiO7BJVq1pbuJWJHFxOR7L8Hv3ZVkSG4eNC8DdwAmDHYu/wadfw0ihKFm2gKDcLHp5yz5UQ8PZ8FyDYvgkrvGV0ak4nc40QDJWws621dm01e/INlGKOIStAAsxOityCLv0zm5Vf3+My/YaBvZcB5mGUsPbx8fAgEAAoGAAy0+WanRqwRHXUzt89OsupPXuNNqBlCEqgTqGAt4Nimq6Ur9u2R1KXKXUotxjp71Ubw6JbuUWvJg+5Rmd9RjT0HOUEQF3rvzEepKtaraPhV5ejEIrB+nJWNfGye4yzLdfEXJBGUQzrG+wNe13izfRNXI4dN/6Q5npzqaqv0E1CkCAQACAQACAQACAQACAQA="
+mp.security.public-key="MIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQCghPCWCobG8nTD24juwSVataW7iViRxcTkey/B792VZEhuHjQvA3cAJgx2Lv8GnX8NIoShZtoCg3Cx6ecs+VEPD2fBcg2L4JK7xldGpOJ3ONEAyVsLOttXZtNXvyDZRijiErQALMTorcgi79M5uVX9/jMv2Ggb2XAeZhlLD28fHwIDAQAB"
+mp.zk.namespace=mpush
diff --git a/mpush-cache/src/main/java/com/mpush/cache/redis/manager/ZKRedisClusterManager.java b/mpush-cache/src/main/java/com/mpush/cache/redis/manager/ZKRedisClusterManager.java
@@ -50,7 +50,7 @@ private ZKRedisClusterManager() {
      */
     @Override
     public void init() {
-        Logs.Console.info("begin init redis cluster");
+        Logs.Console.error("begin init redis cluster");
         List<com.mpush.tools.config.data.RedisGroup> groupList = CC.mp.redis.cluster_group;
         if (groupList.size() > 0) {
             if (CC.mp.redis.write_to_zk) {
@@ -73,7 +73,7 @@ public void init() {
             groups.add(RedisGroup.from(node));
         }
         if (groups.isEmpty()) throw new RuntimeException("init redis sever fail groupList is null");
-        Logs.Console.info("init redis cluster success...");
+        Logs.Console.error("init redis cluster success...");
     }
 
     @Override
@@ -119,7 +119,7 @@ public List<RedisServer> hashSet(String key) {
     private void register(List<com.mpush.tools.config.data.RedisGroup> groupList) {
         String data = Jsons.toJson(groupList);
         ZKClient.I.registerPersist(REDIS_SERVER.getRootPath(), data);
-        Logs.Console.info("register redis server group success, group=" + data);
+        Logs.Console.error("register redis server group success, group=" + data);
     }
 
     public void addGroup(RedisGroup group) {

diff --git a/mpush-core/src/main/java/com/mpush/core/handler/AdminHandler.java b/mpush-core/src/main/java/com/mpush/core/handler/AdminHandler.java
@@ -19,19 +19,26 @@
 
 package com.mpush.core.handler;
 
+import com.google.common.base.Strings;
+import com.mpush.api.Service;
 import com.mpush.cache.redis.RedisKey;
 import com.mpush.cache.redis.manager.RedisManager;
-import com.mpush.tools.config.ConfigManager;
+import com.mpush.common.router.RemoteRouter;
+import com.mpush.core.router.RouterCenter;
+import com.mpush.core.server.AdminServer;
 import com.mpush.tools.Jsons;
 import com.mpush.tools.MPushUtil;
+import com.mpush.tools.config.CC;
+import com.mpush.tools.config.ConfigManager;
 import com.mpush.zk.ZKClient;
 import com.mpush.zk.ZKPath;
 import com.mpush.zk.node.ZKServerNode;
+import com.typesafe.config.ConfigRenderOptions;
 import io.netty.channel.*;
-import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.Serializable;
 import java.util.Date;
 import java.util.List;
 
@@ -40,24 +47,37 @@ public final class AdminHandler extends SimpleChannelInboundHandler<String> {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(AdminHandler.class);
 
-    private static final String DOUBLE_END = "\r\n\r\n";
 
     private static final String EOL = "\r\n";
 
+    private static AdminServer adminServer;
+
+    public AdminHandler(AdminServer adminServer) {
+        this.adminServer = adminServer;
+    }
+
     @Override
     protected void messageReceived(ChannelHandlerContext ctx, String request) throws Exception {
-        Command command = Command.getCommand(request);
+        Command command = Command.help;
-        ChannelFuture future = ctx.write(command.handler(request) + DOUBLE_END);
+        String args = null;
-        if (command.equals(Command.QUIT)) {
+        if (request != null) {
+            String[] cmd_args = request.split(" ");
+            command = Command.toCmd(cmd_args[0].trim());
+            if (cmd_args.length == 2) {
+                args = cmd_args[1];
+            }
+        }
+        Object result = command.handler(ctx, args);
+        ChannelFuture future = ctx.writeAndFlush(result + EOL + EOL);
+        if (command == Command.quit) {
             future.addListener(ChannelFutureListener.CLOSE);
         }
-
     }
 
     @Override
     public void channelActive(ChannelHandlerContext ctx) throws Exception {
         ctx.write("welcome to " + MPushUtil.getInetAddress() + "!" + EOL);
-        ctx.write("It is " + new Date() + " now." + DOUBLE_END);
+        ctx.write("It is " + new Date() + " now." + EOL + EOL);
         ctx.flush();
     }
 
@@ -67,38 +87,118 @@ public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
     }
 
     public enum Command {
-        HELP("help") {
+        help {
             @Override
-            public String handler(String request) {
+            public String handler(ChannelHandlerContext ctx, String args) {
                 StringBuilder buf = new StringBuilder();
-                buf.append("Command:" + EOL);
+                buf.append("Option                               Description" + EOL);
-                buf.append("help:display all command." + EOL);
+                buf.append("------                               -----------" + EOL);
-                buf.append("quit:exit checkHealth." + EOL);
+                buf.append("help                                 show help" + EOL);
-                buf.append("scn:statistics connect num." + EOL);
+                buf.append("quit                                 exit console mode" + EOL);
-                buf.append("rcs:remove current server zk info." + EOL);
+                buf.append("shutdown                             stop mpush server" + EOL);
-                buf.append("scs:stop connection server.");
+                buf.append("restart                              restart mpush server" + EOL);
+                buf.append("zk:<redis, cs ,gs>                   query zk node" + EOL);
+                buf.append("count:<conn, online>                 count conn num or online user count" + EOL);
+                buf.append("route:<uid>                          show user route info" + EOL);
+                buf.append("conf:[key]                           show config info" + EOL);
                 return buf.toString();
             }
         },
-        QUIT("quit") {
+        quit {
             @Override
-            public String handler(String request) {
+            public String handler(ChannelHandlerContext ctx, String args) {
                 return "have a good day!";
             }
         },
-        SCN("scn") {
+        shutdown {
+            @Override
+            public String handler(ChannelHandlerContext ctx, String args) {
+                ctx.writeAndFlush("try close connect server...");
+                adminServer.getConnectionServer().stop(new Service.Listener() {
+                    @Override
+                    public void onSuccess(Object... args) {
+                        ctx.writeAndFlush("connect server close success" + EOL);
+                        adminServer.stop(null);//这个一定要在System.exit之前调用，不然jvm 会卡死 @see com.mpush.bootstrap.Main#addHook
+                        System.exit(0);
+                    }
+
+                    @Override
+                    public void onFailure(Throwable cause) {
+                        ctx.writeAndFlush("connect server close failure, msg=" + cause.getLocalizedMessage());
+                    }
+                });
+                return null;
+            }
+        },
+        restart {
+            @Override
+            public String handler(ChannelHandlerContext ctx, String args) {
+                return "unsupported";
+            }
+        },
+        zk {
+            @Override
+            public String handler(ChannelHandlerContext ctx, String args) {
+                switch (args) {
+                    case "redis":
+                        return ZKClient.I.get(ZKPath.REDIS_SERVER.getRootPath());
+                    case "cs":
+                        return getNodeData(ZKPath.CONNECT_SERVER);
+                    case "gs":
+                        return getNodeData(ZKPath.GATEWAY_SERVER);
+
+                }
+                return "[" + args + "] unsupported, try help.";
+            }
+
+            private String getNodeData(ZKPath path) {
+                List<String> rawData = ZKClient.I.getChildrenKeys(path.getRootPath());
+                StringBuilder sb = new StringBuilder();
+                for (String raw : rawData) {
+                    sb.append(ZKClient.I.get(path.getFullPath(raw))).append('\n');
+                }
+                return sb.toString();
+            }
+        },
+        count {
             @Override
-            public String handler(String request) {
+            public Serializable handler(ChannelHandlerContext ctx, String args) {
-                Long value = RedisManager.I.zCard(RedisKey.getUserOnlineKey(MPushUtil.getExtranetAddress()));
+                switch (args) {
-                if (value == null) {
+                    case "conn":
-                    value = 0L;
+                        return adminServer.getConnectionServer().getConnectionManager().getConnections().size();
+                    case "online": {
+                        Long value = RedisManager.I.zCard(RedisKey.getUserOnlineKey(MPushUtil.getExtranetAddress()));
+                        return value == null ? 0 : value;
+                    }
+
                 }
-                return value.toString() + ".";
+                return "[" + args + "] unsupported, try help.";
+            }
+        },
+        route {
+            @Override
+            public String handler(ChannelHandlerContext ctx, String args) {
+                if (Strings.isNullOrEmpty(args)) return "please input userId";
+                RemoteRouter router = RouterCenter.INSTANCE.getRemoteRouterManager().lookup(args);
+                if (router == null) return "user [" + args + "] offline now.";
+                return router.getRouteValue().toString();
             }
         },
-        RCS("rcs") {
+        conf {
             @Override
-            public String handler(String request) {
+            public String handler(ChannelHandlerContext ctx, String args) {
+                if (Strings.isNullOrEmpty(args)) {
+                    return CC.cfg.root().render(ConfigRenderOptions.concise().setFormatted(true));
+                }
+                if (CC.cfg.hasPath(args)) {
+                    return CC.cfg.getAnyRef(args).toString();
+                }
+                return "key [" + args + "] not find in config";
+            }
+        },
+        rcs {
+            @Override
+            public String handler(ChannelHandlerContext ctx, String args) {
 
                 List<String> rawData = ZKClient.I.getChildrenKeys(ZKPath.CONNECT_SERVER.getRootPath());
                 boolean removeSuccess = false;
@@ -121,35 +221,16 @@ public String handler(String request) {
                     return "remove false.";
                 }
             }
-        },
-        SCS("scs") {
-            @Override
-            public String handler(String request) {
-                return "not support now.";
-            }
         };
-        private final String cmd;
-
-        public abstract String handler(String request);
 
-        private Command(String cmd) {
+        public abstract Object handler(ChannelHandlerContext ctx, String args);
-            this.cmd = cmd;
-        }
 
-        public String getCmd() {
+        public static Command toCmd(String cmd) {
-            return cmd;
+            try {
-        }
+                return Command.valueOf(cmd);
-
+            } catch (Exception e) {
-        public static Command getCommand(String request) {
-            if (StringUtils.isNoneEmpty(request)) {
-                for (Command command : Command.values()) {
-                    if (command.getCmd().equals(request)) {
-                        return command;
-                    }
-                }
             }
-            return HELP;
+            return help;
         }
     }
-
 }