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

eventmesh-dashboard-common/pom.xml

@@ -40,6 +40,38 @@
  <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>
+ <!-- storage redis client -->
+ <dependency>
+ <groupId>io.lettuce</groupId>
+ <artifactId>lettuce-core</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <version>4.13.2</version>
+ </dependency>
+ <!-- Unit Test -->
+ <dependency>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-starter-test</artifactId>
+ </dependency>
+ <!-- Meta - nacos client -->
+ <dependency>
+ <groupId>com.alibaba.nacos</groupId>
+ <artifactId>nacos-client</artifactId>
+ <version>${nacos.version}</version>
+ </dependency>
  </dependencies>
 
 </project>

eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/SDK/config/CreateSDKConfig.java → eventmesh-dashboard-common/src/main/java/org/apache/eventmesh/dashboard/common/model/RocketmqProperties.java

@@ -15,12 +15,28 @@
  * limitations under the License.
  */
 
-package org.apache.eventmesh.dashboard.console.function.SDK.config;
+package org.apache.eventmesh.dashboard.common.model;
 
-/**
- * Config to create an SDK client, usually contains an address url.
- */
-public interface CreateSDKConfig {
+import lombok.Data;
+
+@Data
+public class RocketmqProperties {
+
+ private String namesrvAddr;
+
+ private String clusterName;
+
+ private String brokerUrl;
+
+ private String endPoint;
+
+ private int writeQueueNums;
+
+ private int readQueueNums;
+
+ private String accessKey;
+
+ private String secretKey;
 
- String getUniqueKey();
+ private Long requestTimeoutMillis = 10000L;
 }

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

@@ -17,24 +17,37 @@
 
 package org.apache.eventmesh.dashboard.common.model;
 
-import com.fasterxml.jackson.annotation.JsonCreator;
-import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.rocketmq.common.TopicFilterType;
+
+import lombok.Data;
 
 /**
  * One record displayed in 'Topic' page.
  */
 
+@Data
 public class TopicProperties {
 
- public String name;
- public long messageCount;
-
- @JsonCreator
- public TopicProperties(
- @JsonProperty("name") String name,
- @JsonProperty("messageCount") long messageCount) {
- super();
- this.name = name;
- this.messageCount = messageCount;
- }
+ private static final String SEPARATOR = " ";
+
+ public static int defaultReadQueueNums = 16;
+
+ public static int defaultWriteQueueNums = 16;
+
+ private String topicName;
+
+ private int offset;
+
+ private int readQueueNums;
+
+ private int writeQueueNums;
+
+ private int perm;
+
+ private TopicFilterType topicFilterType;
+
+ private int topicSysFlag;
+
+ private boolean order;
+
 }

eventmesh-dashboard-console/pom.xml

@@ -65,13 +65,6 @@
  <scope>runtime</scope>
  </dependency>
 
- <!-- Meta - nacos client -->
- <dependency>
- <groupId>com.alibaba.nacos</groupId>
- <artifactId>nacos-client</artifactId>
- <version>${nacos.version}</version>
- </dependency>
-
  <!-- health check client -->
  <!-- EventMesh SDK -->
 <!-- <dependency>-->
@@ -93,38 +86,6 @@
 <!-- </exclusion>-->
 <!-- </exclusions>-->
 <!-- </dependency>-->
- <!-- storage redis client -->
- <dependency>
- <groupId>io.lettuce</groupId>
- <artifactId>lettuce-core</artifactId>
- </dependency>
- <!-- rocketmq client -->
- <dependency>
- <groupId>org.apache.rocketmq</groupId>
- <artifactId>rocketmq-client</artifactId>
- <version>4.9.4</version>
- </dependency>
- <dependency>
- <groupId>org.apache.rocketmq</groupId>
- <artifactId>rocketmq-tools</artifactId>
- <version>5.2.0</version>
- </dependency>
- <!-- health check client end -->
-
- <!-- Unit Test -->
- <dependency>
- <groupId>org.springframework.boot</groupId>
- <artifactId>spring-boot-starter-test</artifactId>
- <scope>test</scope>
- </dependency>
-
- <!-- TODO: remove junit4 dependency -->
- <dependency>
- <groupId>junit</groupId>
- <artifactId>junit</artifactId>
- <version>4.13.2</version>
- <scope>test</scope>
- </dependency>
  </dependencies>
 
  <build>

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.common.model.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

@@ -25,6 +25,7 @@
 import org.apache.eventmesh.dashboard.console.function.health.callback.HealthCheckCallback;
 import org.apache.eventmesh.dashboard.console.function.health.check.AbstractHealthCheckService;
 import org.apache.eventmesh.dashboard.console.function.health.check.config.HealthCheckObjectConfig;
+import org.apache.eventmesh.dashboard.service.mq.RocketmqTopicService;
 
 import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
 import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
@@ -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;
@@ -54,12 +52,18 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
+import javax.annotation.Resource;
+
+
 import lombok.extern.slf4j.Slf4j;
 
 @Slf4j
 @HealthCheckType(type = HealthCheckTypeConstant.HEALTH_CHECK_TYPE_STORAGE, subType = HealthCheckTypeConstant.HEALTH_CHECK_SUBTYPE_ROCKETMQ_TOPIC)
 public class Rocketmq4TopicCheck extends AbstractHealthCheckService {
 
+ @Resource
+ private RocketmqTopicService rocketmqTopicService;
+
  private RemotingClient remotingClient;
 
  private DefaultMQPushConsumer consumer;
@@ -138,30 +142,19 @@ 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;
- }
+ rocketmqTopicService.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);
  }