Proxy Support For Dashboard

pom.xml

@@ -163,6 +163,21 @@
                 </exclusion>
             </exclusions>
         </dependency>
+        <dependency>
+            <groupId>org.apache.rocketmq</groupId>
+            <artifactId>rocketmq-proxy</artifactId>
+            <version>${rocketmq.version}</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>ch.qos.logback</groupId>
+                    <artifactId>logback-classic</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>ch.qos.logback</groupId>
+                    <artifactId>logback-core</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
         <dependency>
             <groupId>org.apache.rocketmq</groupId>
             <artifactId>rocketmq-namesrv</artifactId>

src/main/java/org/apache/rocketmq/dashboard/config/RMQConfigure.java

@@ -62,6 +62,10 @@ public class RMQConfigure {
 
     private List<String> namesrvAddrs = new ArrayList<>();
 
+    private volatile String proxyAddr;
+
+    private List<String> proxyAddrs = new ArrayList<>();
+
     public String getAccessKey() {
         return accessKey;
     }
@@ -86,6 +90,25 @@ public List<String> getNamesrvAddrs() {
         return namesrvAddrs;
     }
 
+    public List<String> getProxyAddrs() {
+        return this.proxyAddrs;
+    }
+
+    public void setProxyAddrs(List<String> proxyAddrs) {
+        this.proxyAddrs = proxyAddrs;
+        if (CollectionUtils.isNotEmpty(proxyAddrs)) {
+            this.setProxyAddr(proxyAddrs.get(0));
+        }
+    }
+
+    public String getProxyAddr() {
+        return proxyAddr;
+    }
+
+    public void setProxyAddr(String proxyAddr) {
+        this.proxyAddr = proxyAddr;
+    }
+
     public void setNamesrvAddrs(List<String> namesrvAddrs) {
         this.namesrvAddrs = namesrvAddrs;
         if (CollectionUtils.isNotEmpty(namesrvAddrs)) {

src/main/java/org/apache/rocketmq/dashboard/controller/ProxyController.java

@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.dashboard.controller;
+
+import org.apache.rocketmq.client.exception.MQBrokerException;
+import org.apache.rocketmq.dashboard.permisssion.Permission;
+import org.apache.rocketmq.dashboard.service.ProxyService;
+import org.springframework.stereotype.Controller;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestMethod;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.ResponseBody;
+
+import javax.annotation.Resource;
+
+@Controller
+@RequestMapping("/proxy")
+@Permission
+public class ProxyController {
+    @Resource
+    private ProxyService proxyService;
+    @RequestMapping(value = "/homePage.query", method = RequestMethod.GET)
+    @ResponseBody
+    public Object homePage() {
+        return proxyService.getProxyHomePage();
+    }
+
+    @RequestMapping(value = "/proxyDetailConfig.query", method = RequestMethod.GET)
+    @ResponseBody
+    public Object queryProxyDetailConfig(@RequestParam String proxyAddress) throws MQBrokerException {
+        return proxyService.getProxyDetailConfig(proxyAddress);
+    }
+
+    @RequestMapping(value = "/addProxyAddr.do", method = RequestMethod.POST)
+    @ResponseBody
+    public Object addProxyAddr(@RequestParam String newProxyAddr) {
+        proxyService.addProxyAddrList(newProxyAddr);
+        return true;
+    }
+
+    @RequestMapping(value = "/updateProxyAddr.do", method = RequestMethod.POST)
+    @ResponseBody
+    public Object updateProxyAddr(@RequestParam String proxyAddr) {
+        proxyService.updateProxyAddrList(proxyAddr);
+        return true;
+    }
+}

src/main/java/org/apache/rocketmq/dashboard/controller/TopicController.java

@@ -56,6 +56,12 @@ public Object list(@RequestParam(value = "skipSysProcess", required = false) boo
         return topicService.fetchAllTopicList(skipSysProcess, skipRetryAndDlq);
     }
 
+    @RequestMapping(value = "/list.queryTopicType", method = RequestMethod.GET)
+    @ResponseBody
+    public Object listTopicType() {
+        return topicService.examineAllTopicType();
+    }
+
     @RequestMapping(value = "/stats.query", method = RequestMethod.GET)
     @ResponseBody
     public Object stats(@RequestParam String topic) {

src/main/java/org/apache/rocketmq/dashboard/model/request/TopicTypeList.java

@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.dashboard.model.request;
+
+import java.util.List;
+
+public class TopicTypeList {
+    private List<String> topicNameList;
+    private List<String> messageTypeList;
+
+    public List<String> getTopicNameList() {
+        return topicNameList;
+    }
+
+    public void setTopicNameList(List<String> topicNameList) {
+        this.topicNameList = topicNameList;
+    }
+
+    public List<String> getMessageTypeList() {
+        return messageTypeList;
+    }
+
+    public void setMessageTypeList(List<String> messageTypeList) {
+        this.messageTypeList = messageTypeList;
+    }
+
+    public TopicTypeList(List<String> topicNameList, List<String> messageTypeList) {
+        this.topicNameList = topicNameList;
+        this.messageTypeList = messageTypeList;
+    }
+}

src/main/java/org/apache/rocketmq/dashboard/model/request/TopicTypeMeta.java

@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.dashboard.model.request;
+
+public class TopicTypeMeta {
+    private String topicName;
+    private String messageType;
+
+    public String getTopicName() {
+        return topicName;
+    }
+
+    public void setTopicName(String topicName) {
+        this.topicName = topicName;
+    }
+
+    public String getMessageType() {
+        return messageType;
+    }
+
+    public void setMessageType(String messageType) {
+        this.messageType = messageType;
+    }
+}

src/main/java/org/apache/rocketmq/dashboard/service/ProxyService.java

@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.dashboard.service;
+
+import org.apache.rocketmq.client.exception.MQBrokerException;
+import org.apache.rocketmq.proxy.config.ProxyConfig;
+
+import java.util.Map;
+
+public interface ProxyService {
+    ProxyConfig getProxyDetailConfig(String proxyAddr) throws MQBrokerException;
+
+    void addProxyAddrList(String proxyAddr);
+
+    void updateProxyAddrList(String proxyAddr);
+
+    Map<String, Object>  getProxyHomePage();
+}

src/main/java/org/apache/rocketmq/dashboard/service/TopicService.java

@@ -19,6 +19,7 @@
 
 import org.apache.rocketmq.client.producer.SendResult;
 import org.apache.rocketmq.common.TopicConfig;
+import org.apache.rocketmq.dashboard.model.request.TopicTypeList;
 import org.apache.rocketmq.remoting.protocol.admin.TopicStatsTable;
 import org.apache.rocketmq.remoting.protocol.body.GroupList;
 import org.apache.rocketmq.remoting.protocol.body.TopicList;
@@ -31,6 +32,8 @@
 public interface TopicService {
     TopicList fetchAllTopicList(boolean skipSysProcess, boolean skipRetryAndDlq);
 
+    TopicTypeList examineAllTopicType();
+
     TopicStatsTable stats(String topic);
 
     TopicRouteData route(String topic);

src/main/java/org/apache/rocketmq/dashboard/service/client/ProxyAdmin.java

@@ -0,0 +1,24 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.dashboard.service.client;
+
+import org.apache.rocketmq.client.exception.MQBrokerException;
+import org.apache.rocketmq.proxy.config.ProxyConfig;
+
+public interface ProxyAdmin {
+    ProxyConfig examineProxyConfig(String addr) throws MQBrokerException;
+}

src/main/java/org/apache/rocketmq/dashboard/service/client/ProxyAdminImpl.java

@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.dashboard.service.client;
+
+import com.google.common.base.Throwables;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.pool2.impl.GenericObjectPool;
+import org.apache.rocketmq.client.exception.MQBrokerException;
+import org.apache.rocketmq.proxy.config.ProxyConfig;
+import org.apache.rocketmq.remoting.RemotingClient;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
+import org.apache.rocketmq.remoting.protocol.ResponseCode;
+import org.apache.rocketmq.tools.admin.MQAdminExt;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+@Slf4j
+@Service
+public class ProxyAdminImpl implements ProxyAdmin {
+    @Autowired
+    private GenericObjectPool<MQAdminExt> mqAdminExtPool;
+
+    @Override
+    public ProxyConfig examineProxyConfig(String addr) throws MQBrokerException {
+        try {
+            MQAdminInstance.createMQAdmin(mqAdminExtPool);
+            RemotingClient remotingClient = MQAdminInstance.threadLocalRemotingClient();
+            RemotingCommand request = RemotingCommand.createRequestCommand(514, null);
+            RemotingCommand response = null;
+            try {
+                response = remotingClient.invokeSync(addr, request, 3000);
+            } catch (Exception err) {
+                Throwables.throwIfUnchecked(err);
+                throw new RuntimeException(err);
+            }
+            switch (response.getCode()) {
+                case ResponseCode.SUCCESS: {
+                    ProxyConfig proxyConfig = RemotingSerializable.decode(response.getBody(), ProxyConfig.class);
+                    log.info("addr=" + addr + ",proxyConfig=" + proxyConfig);
+                    return proxyConfig;
+                }
+                default:
+                    throw new MQBrokerException(response.getCode(), response.getRemark());
+            }
+        } finally {
+            MQAdminInstance.returnMQAdmin(mqAdminExtPool);
+        }
+    }
+
+}