[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/RocketmqProperties.java

@@ -0,0 +1,42 @@
+/*
+ * 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.model;
+
+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;
+
+ 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-common/src/main/java/org/apache/eventmesh/dashboard/common/util/RocketmqUtils.java

@@ -0,0 +1,111 @@
+/*
+ * 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 org.springframework.beans.BeanUtils;
+
+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);
+ }
+
+ return topicConfig2TopicProperties(topicConfigList);
+ }
+
+ 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);
+ }
+ }
+
+ private List<TopicProperties> topicConfig2TopicProperties(List<TopicConfig> topicConfigList) {
+ ArrayList<TopicProperties> topicPropertiesList = new ArrayList<>();
+ for (TopicConfig topicConfig : topicConfigList) {
+ TopicProperties topicProperties = new TopicProperties();
+ BeanUtils.copyProperties(topicConfig, topicProperties);
+ topicPropertiesList.add(topicProperties);
+ }
+ return topicPropertiesList;
+ }
+
+}

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

@@ -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);
  }