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

eventmesh-dashboard-common/pom.xml

@@ -40,6 +40,17 @@
             <artifactId>fastjson2</artifactId>
             <version>2.0.40</version>
         </dependency>
+        <!-- Event Store -->
+        <dependency>
+            <groupId>org.apache.rocketmq</groupId>
+            <artifactId>rocketmq-client</artifactId>
+            <version>${rocketmq.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.rocketmq</groupId>
+            <artifactId>rocketmq-tools</artifactId>
+            <version>${rocketmq.version}</version>
+        </dependency>
     </dependencies>
 
 </project>

eventmesh-dashboard-common/src/main/java/org/apache/eventmesh/dashboard/common/model/TopicProperties.java

@@ -17,24 +17,17 @@
 
 package org.apache.eventmesh.dashboard.common.model;
 
-import com.fasterxml.jackson.annotation.JsonCreator;
-import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.rocketmq.common.TopicConfig;
+
+import lombok.Data;
 
 /**
  * One record displayed in 'Topic' page.
  */
 
+@Data
 public class TopicProperties {
 
-    public String name;
-    public long messageCount;
+    private TopicConfig rocketmqTopicConfig;
 
-    @JsonCreator
-    public TopicProperties(
-        @JsonProperty("name") String name,
-        @JsonProperty("messageCount") long messageCount) {
-        super();
-        this.name = name;
-        this.messageCount = messageCount;
-    }
 }

eventmesh-dashboard-common/src/main/java/org/apache/eventmesh/dashboard/common/util/RocketmqUtils.java

@@ -0,0 +1,105 @@
+/*
+ * 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.eventmesh.dashboard.common.util;
+
+import org.apache.eventmesh.dashboard.common.model.TopicProperties;
+
+import org.apache.rocketmq.common.TopicConfig;
+import org.apache.rocketmq.common.protocol.RequestCode;
+import org.apache.rocketmq.common.protocol.body.TopicConfigSerializeWrapper;
+import org.apache.rocketmq.common.protocol.header.CreateTopicRequestHeader;
+import org.apache.rocketmq.common.protocol.header.DeleteTopicRequestHeader;
+import org.apache.rocketmq.remoting.CommandCustomHeader;
+import org.apache.rocketmq.remoting.RemotingClient;
+import org.apache.rocketmq.remoting.netty.NettyClientConfig;
+import org.apache.rocketmq.remoting.netty.NettyRemotingClient;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ConcurrentMap;
+
+import lombok.experimental.UtilityClass;
+import lombok.extern.slf4j.Slf4j;
+
+
+@Slf4j
+@UtilityClass
+public class RocketmqUtils {
+
+    private final RemotingClient remotingClient;
+
+    static {
+        NettyClientConfig config = new NettyClientConfig();
+        config.setUseTLS(false);
+        remotingClient = new NettyRemotingClient(config);
+        remotingClient.start();
+    }
+
+
+    public void createTopic(String topicName, String topicFilterTypeName, int perm, String nameServerAddr,
+        int readQueueNums, int writeQueueNums, long requestTimeoutMillis) {
+        try {
+            CreateTopicRequestHeader requestHeader = new CreateTopicRequestHeader();
+            requestHeader.setTopic(topicName);
+            requestHeader.setTopicFilterType(topicFilterTypeName);
+            requestHeader.setReadQueueNums(readQueueNums);
+            requestHeader.setWriteQueueNums(writeQueueNums);
+            requestHeader.setPerm(perm);
+            RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.UPDATE_AND_CREATE_TOPIC, requestHeader);
+            Object result = remotingClient.invokeSync(nameServerAddr, request, requestTimeoutMillis);
+            log.info("rocketmq create topic result:" + result.toString());
+        } catch (Exception e) {
+            log.error("RocketmqTopicCheck init failed when examining topic stats.", e);
+        }
+    }
+
+    public List<TopicProperties> getTopics(String nameServerAddr, long requestTimeoutMillis) {
+        List<TopicConfig> topicConfigList = new ArrayList<>();
+        try {
+            RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_ALL_TOPIC_CONFIG, (CommandCustomHeader) null);
+            RemotingCommand response = remotingClient.invokeSync(nameServerAddr, request, requestTimeoutMillis);
+            TopicConfigSerializeWrapper allTopicConfig = TopicConfigSerializeWrapper.decode(response.getBody(), TopicConfigSerializeWrapper.class);
+            ConcurrentMap<String, TopicConfig> topicConfigTable = allTopicConfig.getTopicConfigTable();
+            topicConfigList = new ArrayList<>(topicConfigTable.values());
+        } catch (Exception e) {
+            log.error("RocketmqTopicCheck init failed when examining topic stats.", e);
+        }
+        List<TopicProperties> topicPropertiesList = new ArrayList<>();
+        for (TopicConfig topicConfig : topicConfigList) {
+            TopicProperties topicProperties = new TopicProperties();
+            topicProperties.setRocketmqTopicConfig(topicConfig);
+            topicPropertiesList.add(topicProperties);
+        }
+        return topicPropertiesList;
+    }
+
+    public void deleteTopic(String topicName, String nameServerAddr, long requestTimeoutMillis) {
+        try {
+            DeleteTopicRequestHeader deleteTopicRequestHeader = new DeleteTopicRequestHeader();
+            deleteTopicRequestHeader.setTopic(topicName);
+            RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.DELETE_TOPIC_IN_NAMESRV, null);
+            Object result = remotingClient.invokeSync(nameServerAddr, request, requestTimeoutMillis);
+
+            log.info("rocketmq delete topic result:" + result.toString());
+        } catch (Exception e) {
+            log.error("RocketmqTopicCheck init failed when examining topic stats.", e);
+        }
+    }
+
+}

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

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

eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/entity/connection/ConnectionEntity.java

@@ -66,8 +66,7 @@ public class ConnectionEntity extends BaseEntity {
     private String sinkType;
 
     /**
-     * The id of the sink.<br>
-     * It can be connectorId or clientId according to the sinkType.
+     * The id of the sink.<br> It can be connectorId or clientId according to the sinkType.
      */
     @Schema(name = "sinkId", description = "connectorId or clientId")
     private Long sinkId;

eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/health/check/config/HealthCheckObjectConfig.java

@@ -17,10 +17,14 @@
 
 package org.apache.eventmesh.dashboard.console.function.health.check.config;
 
+
+import org.apache.eventmesh.dashboard.service.properties.RocketmqProperties;
+
 import java.util.Properties;
 
 import lombok.Data;
 
+
 @Data
 public class HealthCheckObjectConfig {
 
@@ -56,13 +60,6 @@ public class HealthCheckObjectConfig {
 
     private Long requestTimeoutMillis = 100000L;
 
-    private RocketmqConfig rocketmqConfig = new RocketmqConfig();
-
-    @Data
-    public class RocketmqConfig {
+    private RocketmqProperties rocketmqProperties = new RocketmqProperties();
 
-        private String nameServerUrl;
-        private String brokerUrl;
-        private String endPoint;
-    }
 }

eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/health/check/impl/storage/rocketmq4/Rocketmq4BrokerCheck.java

@@ -51,7 +51,7 @@ public Rocketmq4BrokerCheck(HealthCheckObjectConfig healthCheckObjectConfig) {
     public void doCheck(HealthCheckCallback callback) {
         try {
             RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_BROKER_RUNTIME_INFO, null);
-            remotingClient.invokeAsync(getConfig().getRocketmqConfig().getBrokerUrl(), request, getConfig().getRequestTimeoutMillis(),
+            remotingClient.invokeAsync(getConfig().getRocketmqProperties().getBrokerUrl(), request, getConfig().getRequestTimeoutMillis(),
                 new InvokeCallback() {
                     @Override
                     public void operationComplete(ResponseFuture responseFuture) {
@@ -71,7 +71,7 @@ public void operationComplete(ResponseFuture responseFuture) {
 
     @Override
     public void init() {
-        if (getConfig().getRocketmqConfig().getBrokerUrl() == null || getConfig().getRocketmqConfig().getBrokerUrl().isEmpty()) {
+        if (getConfig().getRocketmqProperties().getBrokerUrl() == null || getConfig().getRocketmqProperties().getBrokerUrl().isEmpty()) {
             throw new IllegalArgumentException("RocketmqCheck failed. BrokerUrl is null.");
         }
 

eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/health/check/impl/storage/rocketmq4/Rocketmq4NameServerCheck.java

@@ -47,7 +47,7 @@ public Rocketmq4NameServerCheck(HealthCheckObjectConfig healthCheckObjectConfig)
     public void doCheck(HealthCheckCallback callback) {
         try {
             RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_NAMESRV_CONFIG, null);
-            remotingClient.invokeAsync(getConfig().getRocketmqConfig().getNameServerUrl(), request, getConfig().getRequestTimeoutMillis(),
+            remotingClient.invokeAsync(getConfig().getRocketmqProperties().getNamesrvAddr(), request, getConfig().getRequestTimeoutMillis(),
                 new InvokeCallback() {
                     @Override
                     public void operationComplete(ResponseFuture responseFuture) {
@@ -67,7 +67,7 @@ public void operationComplete(ResponseFuture responseFuture) {
 
     @Override
     public void init() {
-        if (getConfig().getRocketmqConfig().getNameServerUrl() == null || getConfig().getRocketmqConfig().getNameServerUrl().isEmpty()) {
+        if (getConfig().getRocketmqProperties().getNamesrvAddr() == null || getConfig().getRocketmqProperties().getNamesrvAddr().isEmpty()) {
             throw new RuntimeException("RocketmqNameServerCheck init failed, nameServerUrl is empty");
         }
 

eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/health/check/impl/storage/rocketmq4/Rocketmq4TopicCheck.java

@@ -19,6 +19,7 @@
 
 import static org.apache.rocketmq.client.producer.SendStatus.SEND_OK;
 
+import org.apache.eventmesh.dashboard.common.util.RocketmqUtils;
 import org.apache.eventmesh.dashboard.console.constant.health.HealthCheckTypeConstant;
 import org.apache.eventmesh.dashboard.console.constant.health.HealthConstant;
 import org.apache.eventmesh.dashboard.console.function.health.annotation.HealthCheckType;
@@ -37,14 +38,11 @@
 import org.apache.rocketmq.common.constant.PermName;
 import org.apache.rocketmq.common.message.Message;
 import org.apache.rocketmq.common.message.MessageExt;
-import org.apache.rocketmq.common.protocol.RequestCode;
-import org.apache.rocketmq.common.protocol.header.CreateTopicRequestHeader;
 import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
 import org.apache.rocketmq.remoting.RemotingClient;
 import org.apache.rocketmq.remoting.common.RemotingHelper;
 import org.apache.rocketmq.remoting.netty.NettyClientConfig;
 import org.apache.rocketmq.remoting.netty.NettyRemotingClient;
-import org.apache.rocketmq.remoting.protocol.RemotingCommand;
 
 import java.util.Arrays;
 import java.util.List;
@@ -138,30 +136,18 @@ public void init() {
 
         //TODO there are many functions that can be reused, they should be collected in a util module
         //this function that create topics can be reused
-        try {
-            CreateTopicRequestHeader requestHeader = new CreateTopicRequestHeader();
-            requestHeader.setTopic(HealthConstant.ROCKETMQ_CHECK_TOPIC);
-            requestHeader.setTopicFilterType(TopicFilterType.SINGLE_TAG.name());
-            requestHeader.setReadQueueNums(8);
-            requestHeader.setWriteQueueNums(8);
-            requestHeader.setPerm(PermName.PERM_READ | PermName.PERM_WRITE);
-            RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.UPDATE_AND_CREATE_TOPIC, requestHeader);
-            Object result = remotingClient.invokeSync(getConfig().getRocketmqConfig().getBrokerUrl(), request, getConfig().getRequestTimeoutMillis());
-            log.info(result.toString());
-        } catch (Exception e) {
-            log.error("RocketmqTopicCheck init failed when examining topic stats.", e);
-            return;
-        }
+        RocketmqUtils.createTopic(HealthConstant.ROCKETMQ_CHECK_TOPIC, TopicFilterType.SINGLE_TAG.name(), PermName.PERM_READ | PermName.PERM_WRITE,
+            getConfig().getRocketmqProperties().getBrokerUrl(), 8, 8, getConfig().getRequestTimeoutMillis());
 
         try {
             producer = new DefaultMQProducer(HealthConstant.ROCKETMQ_CHECK_PRODUCER_GROUP);
-            producer.setNamesrvAddr(getConfig().getRocketmqConfig().getNameServerUrl());
+            producer.setNamesrvAddr(getConfig().getRocketmqProperties().getNamesrvAddr());
             producer.setCompressMsgBodyOverHowmuch(16);
             producer.start();
 
             consumer = new DefaultMQPushConsumer(HealthConstant.ROCKETMQ_CHECK_CONSUMER_GROUP);
             consumer.setMessageModel(MessageModel.CLUSTERING);
-            consumer.setNamesrvAddr(getConfig().getRocketmqConfig().getNameServerUrl());
+            consumer.setNamesrvAddr(getConfig().getRocketmqProperties().getNamesrvAddr());
             consumer.subscribe(HealthConstant.ROCKETMQ_CHECK_TOPIC, "*");
             consumer.registerMessageListener(new MessageListenerConcurrently() {
                 @Override

eventmesh-dashboard-console/src/test/java/org/apache/eventmesh/dashboard/console/function/health/check/impl/storage/rocketmq4/RocketmqBrokerCheckTest.java

@@ -35,7 +35,7 @@ class RocketmqBrokerCheckTest {
     @BeforeEach
     public void init() {
         HealthCheckObjectConfig config = new HealthCheckObjectConfig();
-        config.getRocketmqConfig().setBrokerUrl("127.0.0.1:10911");
+        config.getRocketmqProperties().setBrokerUrl("127.0.0.1:10911");
         config.setRequestTimeoutMillis(1000L);
         rocketmqCheck = new Rocketmq4BrokerCheck(config);
     }

eventmesh-dashboard-console/src/test/java/org/apache/eventmesh/dashboard/console/function/health/check/impl/storage/rocketmq4/RocketmqNameserverCheckTest.java

@@ -35,7 +35,7 @@ class RocketmqNameserverCheckTest {
     @BeforeEach
     public void init() {
         HealthCheckObjectConfig config = new HealthCheckObjectConfig();
-        config.getRocketmqConfig().setNameServerUrl("127.0.0.1:9876");
+        config.getRocketmqProperties().setNamesrvAddr("127.0.0.1:9876");
         config.setRequestTimeoutMillis(1000L);
         rocketmqCheck = new Rocketmq4NameServerCheck(config);
     }

eventmesh-dashboard-console/src/test/java/org/apache/eventmesh/dashboard/console/function/health/check/impl/storage/rocketmq4/RocketmqTopicCheckTest.java

@@ -35,8 +35,8 @@ class RocketmqTopicCheckTest {
     @BeforeEach
     public void init() throws InterruptedException {
         HealthCheckObjectConfig config = new HealthCheckObjectConfig();
-        config.getRocketmqConfig().setBrokerUrl("127.0.0.1:10911");
-        config.getRocketmqConfig().setNameServerUrl("127.0.0.1:9876");
+        config.getRocketmqProperties().setBrokerUrl("127.0.0.1:10911");
+        config.getRocketmqProperties().setNamesrvAddr("127.0.0.1:9876");
         config.setRequestTimeoutMillis(1000000L);
         rocketmqCheck = new Rocketmq4TopicCheck(config);
     }

eventmesh-dashboard-core/pom.xml

@@ -58,12 +58,5 @@
             <artifactId>nacos-client</artifactId>
             <version>2.2.4</version>
         </dependency>
-
-        <!-- Event Store -->
-        <dependency>
-            <groupId>org.apache.rocketmq</groupId>
-            <artifactId>rocketmq-client</artifactId>
-            <version>4.9.7</version>
-        </dependency>
     </dependencies>
 </project>