merge branch

Author: allenzhu

rocketmq-tools/src/main/java/com/alibaba/rocketmq/tools/command/consumer/ConsumerStatusSubCommand.java

@@ -25,6 +25,7 @@
 import com.alibaba.rocketmq.common.protocol.body.Connection;
 import com.alibaba.rocketmq.common.protocol.body.ConsumerConnection;
 import com.alibaba.rocketmq.common.protocol.body.ConsumerRunningInfo;
+import com.alibaba.rocketmq.remoting.RPCHook;
 import com.alibaba.rocketmq.tools.admin.DefaultMQAdminExt;
 import com.alibaba.rocketmq.tools.command.SubCommand;
 
@@ -64,8 +65,8 @@ public Options buildCommandlineOptions(Options options) {
 
 
     @Override
-    public void execute(CommandLine commandLine, Options options) {
-        DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt();
+    public void execute(CommandLine commandLine, Options options, RPCHook rpcHook) {
+        DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt(rpcHook);
 
         defaultMQAdminExt.setInstanceName(Long.toString(System.currentTimeMillis()));
 

rocketmq-tools/src/main/java/com/alibaba/rocketmq/tools/command/consumer/StartMonitoringSubCommand.java

@@ -20,6 +20,7 @@
 import org.slf4j.Logger;
 
 import com.alibaba.rocketmq.client.log.ClientLogger;
+import com.alibaba.rocketmq.remoting.RPCHook;
 import com.alibaba.rocketmq.tools.command.SubCommand;
 import com.alibaba.rocketmq.tools.monitor.DefaultMonitorListener;
 import com.alibaba.rocketmq.tools.monitor.MonitorConfig;
@@ -55,10 +56,10 @@ public Options buildCommandlineOptions(Options options) {
 
 
     @Override
-    public void execute(CommandLine commandLine, Options options) {
+    public void execute(CommandLine commandLine, Options options, RPCHook rpcHook) {
         try {
             MonitorService monitorService =
-                    new MonitorService(new MonitorConfig(), new DefaultMonitorListener());
+                    new MonitorService(new MonitorConfig(), new DefaultMonitorListener(), rpcHook);
 
             monitorService.start();
         }

rocketmq-tools/src/main/java/com/alibaba/rocketmq/tools/monitor/MonitorService.java

@@ -28,6 +28,7 @@
 import com.alibaba.rocketmq.common.protocol.body.ConsumerConnection;
 import com.alibaba.rocketmq.common.protocol.body.TopicList;
 import com.alibaba.rocketmq.common.protocol.topic.OffsetMovedEvent;
+import com.alibaba.rocketmq.remoting.RPCHook;
 import com.alibaba.rocketmq.remoting.exception.RemotingException;
 import com.alibaba.rocketmq.tools.admin.DefaultMQAdminExt;
 
@@ -41,17 +42,18 @@ public class MonitorService {
 
     private final MonitorListener monitorListener;
 
-    private final DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt();
+    private final DefaultMQAdminExt defaultMQAdminExt;
     private final DefaultMQPullConsumer defaultMQPullConsumer = new DefaultMQPullConsumer(
         MixAll.TOOLS_CONSUMER_GROUP);
     private final DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer(
         MixAll.MONITOR_CONSUMER_GROUP);
 
 
-    public MonitorService(MonitorConfig monitorConfig, MonitorListener monitorListener) {
+    public MonitorService(MonitorConfig monitorConfig, MonitorListener monitorListener, RPCHook rpcHook) {
         this.monitorConfig = monitorConfig;
         this.monitorListener = monitorListener;
 
+        this.defaultMQAdminExt = new DefaultMQAdminExt(rpcHook);
         this.defaultMQAdminExt.setInstanceName(instanceName());
         this.defaultMQAdminExt.setNamesrvAddr(monitorConfig.getNamesrvAddr());
 
@@ -263,8 +265,13 @@ private void computeUndoneMsgs(final UndoneMsgs undoneMsgs, final ConsumeStats c
 
 
     public static void main(String[] args) throws MQClientException {
-        MonitorService monitorService = new MonitorService(new MonitorConfig(), new DefaultMonitorListener());
+        main0(args, null);
+    }
+
 
+    public static void main0(String[] args, RPCHook rpcHook) throws MQClientException {
+        MonitorService monitorService =
+                new MonitorService(new MonitorConfig(), new DefaultMonitorListener(), rpcHook);
         monitorService.start();
     }
 }