[ISSUE #45] Implement methods from storage-plugin.admin(rocketmq)

eventmesh-dashboard-core/pom.xml

@@ -101,6 +101,16 @@
             <artifactId>rocketmq-client</artifactId>
             <version>4.9.7</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.rocketmq</groupId>
+            <artifactId>rocketmq-tools</artifactId>
+            <version>4.9.5</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.rocketmq</groupId>
+            <artifactId>rocketmq-remoting</artifactId>
+            <version>4.9.5</version>
+        </dependency>
 
         <!-- Database -->
         <dependency>

eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/controller/TopicController.java

@@ -58,23 +58,23 @@ public ResponseEntity<Object> preflight() {
     }
 
     @CrossOrigin
-    @GetMapping
-    public Result<List<TopicProperties>> getList() {
-        List<TopicProperties> topicList = topicService.getTopic();
+    @GetMapping("/getList")
+    public Result<List<TopicProperties>> getList() throws Exception {
+        List<TopicProperties> topicList = topicService.getTopics();
         return Result.success(topicList);
     }
 
     @CrossOrigin
-    @PostMapping
-    public Result<Object> create(@RequestBody CreateTopicRequest createTopicRequest) {
+    @PostMapping("/create")
+    public Result<Object> create(@RequestBody CreateTopicRequest createTopicRequest) throws Exception {
         String topicName = createTopicRequest.getName();
         topicService.createTopic(topicName);
         return Result.success();
     }
 
     @CrossOrigin
-    @DeleteMapping
-    public Result<Object> delete(@RequestBody DeleteTopicRequest deleteTopicRequest) {
+    @DeleteMapping("/delete")
+    public Result<Object> delete(@RequestBody DeleteTopicRequest deleteTopicRequest) throws Exception {
         String topicName = deleteTopicRequest.getName();
         topicService.deleteTopic(topicName);
         return Result.success();

eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/service/TopicService.java

@@ -27,12 +27,9 @@
 
 public interface TopicService {
 
-    /**
-     * TODO rename to getTopics after being implemented
-     */
-    List<TopicProperties> getTopic();
+    List<TopicProperties> getTopics() throws Exception;
 
-    void createTopic(String topicName);
+    void createTopic(String topicName) throws Exception;
 
-    void deleteTopic(String topicName);
+    void deleteTopic(String topicName) throws Exception;
 }

eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/service/store/RocketmqTopicService.java

@@ -21,38 +21,127 @@
 import org.apache.eventmesh.dashboard.core.model.TopicProperties;
 import org.apache.eventmesh.dashboard.core.service.TopicService;
 
+import org.apache.commons.lang3.StringUtils;
+import org.apache.rocketmq.acl.common.AclClientRPCHook;
+import org.apache.rocketmq.acl.common.SessionCredentials;
+import org.apache.rocketmq.common.TopicConfig;
+import org.apache.rocketmq.common.admin.TopicOffset;
+import org.apache.rocketmq.common.admin.TopicStatsTable;
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.remoting.RPCHook;
+import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
+import org.apache.rocketmq.tools.command.CommandUtil;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Set;
+import java.util.UUID;
 
 import org.springframework.stereotype.Service;
 
 import lombok.extern.slf4j.Slf4j;
 
-/**
- * TODO implement methods from storage-plugin.admin
- */
 
 @Slf4j
 @Service
 public class RocketmqTopicService implements TopicService {
 
     AdminProperties adminProperties;
 
+    private final RPCHook rpcHook;
+
+    protected String nameServerAddr;
+
+    protected String clusterName;
+
+    private int numOfQueue = 4;
+    private int queuePermission = 6;
+
+    private String accessKey;
+
+    private String secretKey;
+
+
+
     public RocketmqTopicService(AdminProperties adminProperties) {
         this.adminProperties = adminProperties;
+        this.nameServerAddr = adminProperties.getStore().getRocketmq().getNamesrvAddr();
+        this.clusterName = adminProperties.getStore().getRocketmq().getClusterName();
+        this.accessKey = adminProperties.getStore().getRocketmq().getAccessKey();
+        this.secretKey = adminProperties.getStore().getRocketmq().getSecretKey();
+        this.rpcHook = new AclClientRPCHook(new SessionCredentials(accessKey, secretKey));
     }
 
-    @Override
-    public List<TopicProperties> getTopic() {
-        return null;
+    private DefaultMQAdminExt createMQAdminExt() {
+        DefaultMQAdminExt adminExt = new DefaultMQAdminExt(rpcHook);
+        String groupId = UUID.randomUUID().toString();
+        adminExt.setAdminExtGroup("admin_ext_group-" + groupId);
+        adminExt.setNamesrvAddr(nameServerAddr);
+        return adminExt;
     }
 
     @Override
-    public void createTopic(String topicName) {
+    public List<TopicProperties> getTopics() throws Exception {
+        DefaultMQAdminExt adminExt = createMQAdminExt();
+        try {
+            List<TopicProperties> result = new ArrayList<>();
 
+            adminExt.start();
+            Set<String> topicList = adminExt.fetchAllTopicList().getTopicList();
+            for (String topic : topicList) {
+                long messageCount = 0;
+                TopicStatsTable topicStats = adminExt.examineTopicStats(topic);
+                HashMap<MessageQueue, TopicOffset> offsetTable = topicStats.getOffsetTable();
+                for (TopicOffset topicOffset : offsetTable.values()) {
+                    messageCount += topicOffset.getMaxOffset() - topicOffset.getMinOffset();
+                }
+                result.add(new TopicProperties(
+                    topic, messageCount));
+            }
+
+            result.sort(Comparator.comparing(t -> t.name));
+            return result;
+        } finally {
+            adminExt.shutdown();
+        }
     }
 
     @Override
-    public void deleteTopic(String topicName) {
+    public void createTopic(String topicName) throws Exception {
+        if (StringUtils.isBlank(topicName)) {
+            throw new Exception("Topic name can not be blank");
+        }
+        DefaultMQAdminExt adminExt = createMQAdminExt();
+        try {
+            adminExt.start();
+            Set<String> brokerAddress = CommandUtil.fetchMasterAddrByClusterName(adminExt, clusterName);
+            for (String masterAddress : brokerAddress) {
+                TopicConfig topicConfig = new TopicConfig();
+                topicConfig.setTopicName(topicName);
+                topicConfig.setReadQueueNums(numOfQueue);
+                topicConfig.setWriteQueueNums(numOfQueue);
+                topicConfig.setPerm(queuePermission);
+                adminExt.createAndUpdateTopicConfig(masterAddress, topicConfig);
+            }
+        } finally {
+            adminExt.shutdown();
+        }
+    }
 
+    @Override
+    public void deleteTopic(String topicName) throws Exception {
+        if (StringUtils.isBlank(topicName)) {
+            throw new Exception("Topic name can not be blank.");
+        }
+        DefaultMQAdminExt adminExt = createMQAdminExt();
+        try {
+            adminExt.start();
+            Set<String> brokerAddress = CommandUtil.fetchMasterAddrByClusterName(adminExt, clusterName);
+            adminExt.deleteTopicInBroker(brokerAddress, topicName);
+        } finally {
+            adminExt.shutdown();
+        }
     }
 }