This is an automated email from the ASF dual-hosted git repository.
mikexue pushed a commit to branch eventmesh-connector
in repository https://gitbox.apache.org/repos/asf/eventmesh.git
The following commit(s) were added to refs/heads/eventmesh-connector by this
push:
new e81a1f425 [ISSUE #3913] Add Source Connector RocketMQ Module (#3914)
e81a1f425 is described below
commit e81a1f4251e4c2a931f2416b2e306ed4cdcdef5f
Author: mike_xwm <[email protected]>
AuthorDate: Fri May 12 17:30:49 2023 +0800
[ISSUE #3913] Add Source Connector RocketMQ Module (#3914)
---
.../source-connector-rocketmq/build.gradle | 41 ++++++++
.../source-connector-rocketmq/gradle.properties | 17 ++++
.../connector/rocketmq/EventMeshTestUtils.java | 113 +++++++++++++++++++++
.../connector/rocketmq/RocketMQSourceWorker.java | 60 +++++++++++
.../source/connector/rocketmq/UtilsConstants.java | 52 ++++++++++
.../rocketmq/config/RocketMQSourceConfig.java | 63 ++++++++++++
.../connector/RocketMQSourceConnector.java | 107 +++++++++++++++++++
settings.gradle | 1 +
8 files changed, 454 insertions(+)
diff --git a/eventmesh-connectors/source-connector-rocketmq/build.gradle
b/eventmesh-connectors/source-connector-rocketmq/build.gradle
new file mode 100644
index 000000000..180094cf2
--- /dev/null
+++ b/eventmesh-connectors/source-connector-rocketmq/build.gradle
@@ -0,0 +1,41 @@
+/*
+ * 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.
+ */
+
+List rocketmq = [
+ "org.apache.rocketmq:rocketmq-client:$rocketmq_version",
+ "org.apache.rocketmq:rocketmq-broker:$rocketmq_version",
+ "org.apache.rocketmq:rocketmq-common:$rocketmq_version",
+ "org.apache.rocketmq:rocketmq-store:$rocketmq_version",
+ "org.apache.rocketmq:rocketmq-namesrv:$rocketmq_version",
+ "org.apache.rocketmq:rocketmq-tools:$rocketmq_version",
+ "org.apache.rocketmq:rocketmq-remoting:$rocketmq_version",
+ "org.apache.rocketmq:rocketmq-logging:$rocketmq_version",
+ "org.apache.rocketmq:rocketmq-test:$rocketmq_version",
+ "org.apache.rocketmq:rocketmq-srvutil:$rocketmq_version",
+ "org.apache.rocketmq:rocketmq-filter:$rocketmq_version",
+ "org.apache.rocketmq:rocketmq-acl:$rocketmq_version",
+ "org.apache.rocketmq:rocketmq-srvutil:$rocketmq_version",
+
+]
+
+dependencies {
+ implementation project(":eventmesh-connectors:eventmesh-connector-api")
+ implementation project(":eventmesh-sdk-java")
+ implementation rocketmq
+ compileOnly 'org.projectlombok:lombok'
+ annotationProcessor 'org.projectlombok:lombok'
+}
diff --git a/eventmesh-connectors/source-connector-rocketmq/gradle.properties
b/eventmesh-connectors/source-connector-rocketmq/gradle.properties
new file mode 100644
index 000000000..89a6e4612
--- /dev/null
+++ b/eventmesh-connectors/source-connector-rocketmq/gradle.properties
@@ -0,0 +1,17 @@
+#
+# 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.
+#
+rocketmq_version=4.9.5
\ No newline at end of file
diff --git
a/eventmesh-connectors/source-connector-rocketmq/src/main/java/org/apache/eventmesh/source/connector/rocketmq/EventMeshTestUtils.java
b/eventmesh-connectors/source-connector-rocketmq/src/main/java/org/apache/eventmesh/source/connector/rocketmq/EventMeshTestUtils.java
new file mode 100644
index 000000000..8e96ef01c
--- /dev/null
+++
b/eventmesh-connectors/source-connector-rocketmq/src/main/java/org/apache/eventmesh/source/connector/rocketmq/EventMeshTestUtils.java
@@ -0,0 +1,113 @@
+/*
+ * 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.source.connector.rocketmq;
+
+import static
org.apache.eventmesh.common.protocol.tcp.Command.RESPONSE_TO_SERVER;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.UUID;
+import java.util.concurrent.ThreadLocalRandom;
+import org.apache.eventmesh.client.tcp.common.EventMeshCommon;
+import org.apache.eventmesh.client.tcp.common.MessageUtils;
+import org.apache.eventmesh.common.protocol.tcp.EventMeshMessage;
+import org.apache.eventmesh.common.protocol.tcp.Header;
+import org.apache.eventmesh.common.protocol.tcp.Package;
+import org.apache.eventmesh.common.protocol.tcp.UserAgent;
+import org.apache.eventmesh.common.utils.JsonUtils;
+
+public class EventMeshTestUtils {
+
+ private static final int SEQ_LENGTH = 10;
+
+ private static final String ASYNC_MSG_BODY = "testAsyncMessage";
+
+ private static final String DEFAULT_TTL_MS = "30000";
+
+ // generate pub-client
+ public static UserAgent generateClient1() {
+ final UserAgent agent = UserAgent.builder()
+ .env(UtilsConstants.ENV)
+ .host(UtilsConstants.HOST)
+ .password(generateRandomString(UtilsConstants.PASSWORD_LENGTH))
+ .username(UtilsConstants.USER_NAME)
+ .group(UtilsConstants.GROUP)
+ .path(UtilsConstants.PATH)
+ .port(UtilsConstants.PORT_1)
+ .subsystem(UtilsConstants.SUB_SYSTEM_1)
+ .pid(UtilsConstants.PID_1)
+ .version(UtilsConstants.VERSION)
+ .idc(UtilsConstants.IDC)
+ .build();
+ return MessageUtils.generatePubClient(agent);
+ }
+
+ // generate sub-client
+ public static UserAgent generateClient2() {
+ final UserAgent agent = UserAgent.builder()
+ .env(UtilsConstants.ENV)
+ .host(UtilsConstants.HOST)
+ .password(generateRandomString(UtilsConstants.PASSWORD_LENGTH))
+ .username(UtilsConstants.USER_NAME)
+ .group(UtilsConstants.GROUP)
+ .path(UtilsConstants.PATH)
+ .port(UtilsConstants.PORT_2)
+ .subsystem(UtilsConstants.SUB_SYSTEM_2)
+ .pid(UtilsConstants.PID_2)
+ .version(UtilsConstants.VERSION)
+ .idc(UtilsConstants.IDC)
+ .build();
+ return MessageUtils.generateSubClient(agent);
+ }
+
+ public static Package rrResponse(final EventMeshMessage request) {
+ final Package msg = new Package();
+ msg.setHeader(new Header(RESPONSE_TO_SERVER, 0, null,
generateRandomString(SEQ_LENGTH)));
+ msg.setBody(request);
+ return msg;
+ }
+
+ private static String generateRandomString(final int length) {
+ final StringBuilder builder = new StringBuilder(length);
+ for (int i = 0; i < length; i++) {
+ builder.append((char) ThreadLocalRandom.current().nextInt(48, 57));
+ }
+ return builder.toString();
+ }
+
+ public static CloudEvent generateCloudEventV1(String destination, String
message) {
+ final Map<String, String> content = new HashMap<>();
+ content.put(UtilsConstants.CONTENT, message);
+
+ return CloudEventBuilder.v1()
+ .withId(UUID.randomUUID().toString())
+ .withSubject(destination)
+ .withSource(URI.create("/"))
+ .withDataContentType("application/cloudevents+json")
+ .withType(EventMeshCommon.CLOUD_EVENTS_PROTOCOL_NAME)
+
.withData(Objects.requireNonNull(JsonUtils.toJSONString(content)).getBytes(StandardCharsets.UTF_8))
+ .withExtension(UtilsConstants.TTL, DEFAULT_TTL_MS)
+ .build();
+ }
+
+}
diff --git
a/eventmesh-connectors/source-connector-rocketmq/src/main/java/org/apache/eventmesh/source/connector/rocketmq/RocketMQSourceWorker.java
b/eventmesh-connectors/source-connector-rocketmq/src/main/java/org/apache/eventmesh/source/connector/rocketmq/RocketMQSourceWorker.java
new file mode 100644
index 000000000..142c68e95
--- /dev/null
+++
b/eventmesh-connectors/source-connector-rocketmq/src/main/java/org/apache/eventmesh/source/connector/rocketmq/RocketMQSourceWorker.java
@@ -0,0 +1,60 @@
+package org.apache.eventmesh.source.connector.rocketmq;
+
+import io.cloudevents.CloudEvent;
+import java.util.List;
+import org.apache.eventmesh.client.tcp.EventMeshTCPClient;
+import org.apache.eventmesh.client.tcp.EventMeshTCPClientFactory;
+import org.apache.eventmesh.client.tcp.conf.EventMeshTCPClientConfig;
+import org.apache.eventmesh.common.protocol.tcp.UserAgent;
+import org.apache.eventmesh.connector.api.data.ConnectRecord;
+import
org.apache.eventmesh.source.connector.rocketmq.config.RocketMQSourceConfig;
+import
org.apache.eventmesh.source.connector.rocketmq.connector.RocketMQSourceConnector;
+
+public class RocketMQSourceWorker {
+
+ public static final String SOURCE_CONSUMER_GROUP =
"DEFAULT-CONSUMER-GROUP";
+ public static final String SOURCE_CONNECT_NAMESRVADDR = "127.0.0.1:9877";
+ public static final String SOURCE_TOPIC = "TopicTest";
+
+ public static final String DESTINATION = "SourceTopic";
+
+
+ public static void main(String[] args) throws Exception {
+
+ UserAgent userAgent = EventMeshTestUtils.generateClient1();
+ EventMeshTCPClientConfig eventMeshTcpClientConfig =
EventMeshTCPClientConfig.builder()
+ .host("127.0.0.1")
+ .port(10002)
+ .userAgent(userAgent)
+ .build();
+
+ final EventMeshTCPClient<CloudEvent> client =
+
EventMeshTCPClientFactory.createEventMeshTCPClient(eventMeshTcpClientConfig,
CloudEvent.class);
+
+ client.init();
+
+ RocketMQSourceConnector rocketMQSourceConnector = new
RocketMQSourceConnector();
+
+ RocketMQSourceConfig rocketMQSourceConfig = new RocketMQSourceConfig();
+
+ rocketMQSourceConfig.setSourceNameserver(SOURCE_CONNECT_NAMESRVADDR);
+ rocketMQSourceConfig.setSourceTopic(SOURCE_TOPIC);
+ rocketMQSourceConfig.setSourceGroup(SOURCE_CONSUMER_GROUP);
+
+ rocketMQSourceConnector.init(rocketMQSourceConfig);
+
+ rocketMQSourceConnector.start();
+
+ while(true) {
+ List<ConnectRecord> connectorRecordList =
rocketMQSourceConnector.poll();
+ for(ConnectRecord connectRecord : connectorRecordList) {
+ // todo:connectorRecord 转换 cloudEvents
+ CloudEvent event =
EventMeshTestUtils.generateCloudEventV1(connectRecord.getExtension("topic"),
connectRecord.getData().toString());
+ client.publish(event, 3000);
+ Thread.sleep(500);
+ }
+ }
+
+ }
+
+}
diff --git
a/eventmesh-connectors/source-connector-rocketmq/src/main/java/org/apache/eventmesh/source/connector/rocketmq/UtilsConstants.java
b/eventmesh-connectors/source-connector-rocketmq/src/main/java/org/apache/eventmesh/source/connector/rocketmq/UtilsConstants.java
new file mode 100644
index 000000000..29cfa028e
--- /dev/null
+++
b/eventmesh-connectors/source-connector-rocketmq/src/main/java/org/apache/eventmesh/source/connector/rocketmq/UtilsConstants.java
@@ -0,0 +1,52 @@
+/*
+ * 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.source.connector.rocketmq;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class UtilsConstants {
+
+ public static final String ENV = "test";
+ public static final String HOST = "localhost";
+ public static final Integer PASSWORD_LENGTH = 8;
+ public static final String USER_NAME = "PU4283";
+ public static final String GROUP = "EventmeshTestGroup";
+ public static final String PATH = "/data/app/umg_proxy";
+ public static final Integer PORT_1 = 8362;
+ public static final Integer PORT_2 = 9362;
+ public static final String SUB_SYSTEM_1 = "5023";
+ public static final String SUB_SYSTEM_2 = "5017";
+ public static final Integer PID_1 = 32_893;
+ public static final Integer PID_2 = 42_893;
+ public static final String VERSION = "2.0.11";
+ public static final String IDC = "FT";
+ /**
+ * PROPERTY KEY NAME .
+ */
+ public static final String MSG_TYPE = "msgtype";
+ public static final String TTL = "ttl";
+ public static final String KEYS = "keys";
+ public static final String REPLY_TO = "replyto";
+ public static final String PROPERTY_MESSAGE_REPLY_TO =
"propertymessagereplyto";
+ public static final String CONTENT = "content";
+
+
+}
diff --git
a/eventmesh-connectors/source-connector-rocketmq/src/main/java/org/apache/eventmesh/source/connector/rocketmq/config/RocketMQSourceConfig.java
b/eventmesh-connectors/source-connector-rocketmq/src/main/java/org/apache/eventmesh/source/connector/rocketmq/config/RocketMQSourceConfig.java
new file mode 100644
index 000000000..4b5f54829
--- /dev/null
+++
b/eventmesh-connectors/source-connector-rocketmq/src/main/java/org/apache/eventmesh/source/connector/rocketmq/config/RocketMQSourceConfig.java
@@ -0,0 +1,63 @@
+/*
+ * 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.source.connector.rocketmq.config;
+
+import org.apache.eventmesh.connector.api.config.SourceConfig;
+
+public class RocketMQSourceConfig extends SourceConfig {
+
+ String connectorName;
+
+ String sourceNameserver;
+
+ String sourceTopic;
+
+ String sourceGroup;
+
+ public String getConnectorName() {
+ return connectorName;
+ }
+
+ public void setConnectorName(String connectorName) {
+ this.connectorName = connectorName;
+ }
+
+ public String getSourceNameserver() {
+ return sourceNameserver;
+ }
+
+ public void setSourceNameserver(String sourceNameserver) {
+ this.sourceNameserver = sourceNameserver;
+ }
+
+ public String getSourceTopic() {
+ return sourceTopic;
+ }
+
+ public void setSourceTopic(String sourceTopic) {
+ this.sourceTopic = sourceTopic;
+ }
+
+ public String getSourceGroup() {
+ return sourceGroup;
+ }
+
+ public void setSourceGroup(String sourceGroup) {
+ this.sourceGroup = sourceGroup;
+ }
+}
diff --git
a/eventmesh-connectors/source-connector-rocketmq/src/main/java/org/apache/eventmesh/source/connector/rocketmq/connector/RocketMQSourceConnector.java
b/eventmesh-connectors/source-connector-rocketmq/src/main/java/org/apache/eventmesh/source/connector/rocketmq/connector/RocketMQSourceConnector.java
new file mode 100644
index 000000000..751e6ef75
--- /dev/null
+++
b/eventmesh-connectors/source-connector-rocketmq/src/main/java/org/apache/eventmesh/source/connector/rocketmq/connector/RocketMQSourceConnector.java
@@ -0,0 +1,107 @@
+/*
+ * 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.source.connector.rocketmq.connector;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.eventmesh.connector.api.config.Config;
+import org.apache.eventmesh.connector.api.data.ConnectRecord;
+import org.apache.eventmesh.connector.api.data.RecordOffset;
+import org.apache.eventmesh.connector.api.data.RecordPartition;
+import org.apache.eventmesh.connector.api.source.Source;
+import
org.apache.eventmesh.source.connector.rocketmq.config.RocketMQSourceConfig;
+import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
+import org.apache.rocketmq.common.message.MessageExt;
+
+public class RocketMQSourceConnector implements Source {
+
+ private RocketMQSourceConfig sourceConfig;
+
+ private final DefaultLitePullConsumer consumer = new
DefaultLitePullConsumer();
+
+ @Override
+ public Class<? extends Config> configClass() {
+ return RocketMQSourceConfig.class;
+ }
+
+ @Override
+ public void init(Config config) throws Exception {
+ // init config for rocketmq source connector
+ this.sourceConfig = (RocketMQSourceConfig) config;
+ consumer.setConsumerGroup(sourceConfig.getSourceGroup());
+ consumer.setNamesrvAddr(sourceConfig.getSourceNameserver());
+ }
+
+ @Override
+ public void start() throws Exception {
+ consumer.subscribe(sourceConfig.getSourceTopic(), "*");
+ consumer.start();
+ }
+
+ @Override
+ public void commit(ConnectRecord record) {
+
+ }
+
+ @Override
+ public String name() {
+ return this.sourceConfig.getConnectorName();
+ }
+
+ @Override
+ public void stop() {
+ consumer.unsubscribe(sourceConfig.getSourceTopic());
+ consumer.shutdown();
+ }
+
+ @Override
+ public List<ConnectRecord> poll() {
+ List<MessageExt> messageExts = consumer.poll();
+ List<ConnectRecord> connectRecords = new
ArrayList<>(messageExts.size());
+ System.out.printf("%s Receive New Messages: %s %n",
Thread.currentThread().getName(), messageExts);
+ for (MessageExt messageExt : messageExts) {
+ Long timestamp = System.currentTimeMillis();
+ byte[] body = messageExt.getBody();
+ String bodyStr = new String(body, StandardCharsets.UTF_8);
+ RecordPartition recordPartition =
convertToRecordPartition(messageExt.getTopic(),
+ messageExt.getBrokerName(), messageExt.getQueueId());
+ RecordOffset recordOffset =
convertToRecordOffset(messageExt.getQueueOffset());
+ ConnectRecord connectRecord = new ConnectRecord(recordPartition,
recordOffset, timestamp, bodyStr);
+ connectRecord.addExtension("topic", messageExt.getTopic());
+ connectRecords.add(connectRecord);
+ }
+ return connectRecords;
+ }
+
+ public static RecordOffset convertToRecordOffset(Long offset) {
+ Map<String, String> offsetMap = new HashMap<>();
+ offsetMap.put("queueOffset", offset + "");
+ return new RecordOffset(offsetMap);
+ }
+
+ public static RecordPartition convertToRecordPartition(String topic,
String brokerName, int queueId) {
+ Map<String, String> map = new HashMap<>();
+ map.put("topic", topic);
+ map.put("brokerName", brokerName);
+ map.put("queueId", queueId + "");
+ return new RecordPartition(map);
+ }
+}
diff --git a/settings.gradle b/settings.gradle
index d6dfaa6a1..722fa0ab7 100644
--- a/settings.gradle
+++ b/settings.gradle
@@ -24,6 +24,7 @@ include 'eventmesh-starter'
include 'eventmesh-examples'
include 'eventmesh-spi'
include 'eventmesh-connectors:eventmesh-connector-api'
+include 'eventmesh-connectors:source-connector-rocketmq'
include 'eventmesh-storage-plugin:eventmesh-storage-api'
include 'eventmesh-storage-plugin:eventmesh-storage-standalone'
include 'eventmesh-storage-plugin:eventmesh-storage-kafka'
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
