This is an automated email from the ASF dual-hosted git repository.
wqliang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/eventmesh.git
The following commit(s) were added to refs/heads/master by this push:
new 623146be6 [Feature] Add eventmesh-connector-sdks module
new b95bd8bcb Merge pull request #3976 from
xwm1992/add-eventmesh-connector-sdks
623146be6 is described below
commit 623146be62c1d85a87d97bf246a9a257dd94c004
Author: xwm1992 <[email protected]>
AuthorDate: Fri May 19 18:26:35 2023 +0800
[Feature] Add eventmesh-connector-sdks module
---
build.gradle | 1 +
eventmesh-connector-sdks/README.md | 0
.../build.gradle | 2 +-
.../eventmesh-connector-sdk-java}/build.gradle | 17 +++
.../gradle.properties | 0
.../apache/eventmesh/connector/Application.java | 103 +++++++++++++++
.../eventmesh/connector/ConnectorWorker.java | 7 +
.../org/apache/eventmesh/connector/SinkWorker.java | 145 ++++++++++++++++++++
.../apache/eventmesh/connector/SourceWorker.java | 146 +++++++++++++++++++++
.../eventmesh/connector/api/config/Config.java | 0
.../eventmesh/connector/api/config/Constants.java | 12 +-
.../connector/api/config/PubSubConfig.java | 22 +++-
.../eventmesh/connector/api/config/SinkConfig.java | 6 +
.../connector/api/config/SourceConfig.java | 6 +
.../connector/api/connector/Connector.java | 0
.../connector/api/connector/SinkConnector.java | 0
.../connector/api/connector/SourceConnector.java | 0
.../connector/api/data/ConnectRecord.java | 0
.../connector/api/data/DefaultKeyValue.java | 0
.../eventmesh/connector/api/data/KeyValue.java | 0
.../eventmesh/connector/api/data/RecordOffset.java | 0
.../connector/api/data/RecordPartition.java | 0
.../connector/api/data/RecordPosition.java | 0
.../apache/eventmesh/connector/api/sink/Sink.java | 0
.../eventmesh/connector/api/source/Source.java | 0
.../eventmesh/connector/util/ConfigUtil.java | 69 ++++++++++
.../gradle.properties | 0
settings.gradle | 3 +-
28 files changed, 535 insertions(+), 4 deletions(-)
diff --git a/build.gradle b/build.gradle
index f24508a1f..920c39072 100644
--- a/build.gradle
+++ b/build.gradle
@@ -449,6 +449,7 @@ subprojects {
dependency "com.fasterxml.jackson.core:jackson-databind:2.13.0"
dependency "com.fasterxml.jackson.core:jackson-core:2.13.0"
dependency "com.fasterxml.jackson.core:jackson-annotations:2.13.0"
+ dependency
"com.fasterxml.jackson.dataformat:jackson-dataformat-yaml:2.13.0"
dependency "org.asynchttpclient:async-http-client:2.12.0"
dependency "org.apache.httpcomponents:httpclient:4.5.13"
diff --git a/eventmesh-connector-sdks/README.md
b/eventmesh-connector-sdks/README.md
new file mode 100644
index 000000000..e69de29bb
diff --git a/eventmesh-connectors/eventmesh-connector-api/build.gradle
b/eventmesh-connector-sdks/build.gradle
similarity index 99%
copy from eventmesh-connectors/eventmesh-connector-api/build.gradle
copy to eventmesh-connector-sdks/build.gradle
index 2944f9819..d973dceda 100644
--- a/eventmesh-connectors/eventmesh-connector-api/build.gradle
+++ b/eventmesh-connector-sdks/build.gradle
@@ -13,4 +13,4 @@
* 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.
- */
+ */
\ No newline at end of file
diff --git a/eventmesh-connectors/eventmesh-connector-api/build.gradle
b/eventmesh-connector-sdks/eventmesh-connector-sdk-java/build.gradle
similarity index 55%
rename from eventmesh-connectors/eventmesh-connector-api/build.gradle
rename to eventmesh-connector-sdks/eventmesh-connector-sdk-java/build.gradle
index 2944f9819..a671db906 100644
--- a/eventmesh-connectors/eventmesh-connector-api/build.gradle
+++ b/eventmesh-connector-sdks/eventmesh-connector-sdk-java/build.gradle
@@ -14,3 +14,20 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
+dependencies {
+ api "org.slf4j:slf4j-api"
+ implementation "org.apache.logging.log4j:log4j-api"
+ implementation "org.apache.logging.log4j:log4j-core"
+ implementation "org.apache.logging.log4j:log4j-slf4j-impl"
+
+ implementation "com.fasterxml.jackson.core:jackson-databind"
+ implementation "com.fasterxml.jackson.core:jackson-core"
+ implementation "com.fasterxml.jackson.core:jackson-annotations"
+ implementation "com.fasterxml.jackson.dataformat:jackson-dataformat-yaml"
+
+ implementation project(":eventmesh-sdk-java")
+
+ compileOnly 'org.projectlombok:lombok'
+ annotationProcessor 'org.projectlombok:lombok'
+}
\ No newline at end of file
diff --git a/eventmesh-connectors/eventmesh-connector-api/gradle.properties
b/eventmesh-connector-sdks/eventmesh-connector-sdk-java/gradle.properties
similarity index 100%
copy from eventmesh-connectors/eventmesh-connector-api/gradle.properties
copy to eventmesh-connector-sdks/eventmesh-connector-sdk-java/gradle.properties
diff --git
a/eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/Application.java
b/eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/Application.java
new file mode 100644
index 000000000..913bdd41f
--- /dev/null
+++
b/eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/Application.java
@@ -0,0 +1,103 @@
+/*
+ * 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.connector;
+
+import org.apache.eventmesh.connector.api.config.Config;
+import org.apache.eventmesh.connector.api.config.SinkConfig;
+import org.apache.eventmesh.connector.api.config.SourceConfig;
+import org.apache.eventmesh.connector.api.connector.Connector;
+import org.apache.eventmesh.connector.api.sink.Sink;
+import org.apache.eventmesh.connector.api.source.Source;
+import org.apache.eventmesh.connector.util.ConfigUtil;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class Application {
+
+ public static void run(Class<? extends Connector> clazz) throws Exception {
+ Connector connector;
+ try {
+ connector = clazz.getDeclaredConstructor().newInstance();
+ } catch (Exception e) {
+ log.error("new connector error", e);
+ return;
+ }
+ Config config;
+ try {
+ config = ConfigUtil.parse(connector.configClass());
+ // offset storage, memory default
+// KVStoreFactory.setStoreConfig(config.getStoreConfig());
+ } catch (Exception e) {
+ log.error("parse config error", e);
+ return;
+ }
+ try {
+ connector.init(config);
+ } catch (Exception e) {
+ log.error("connector {} initialize error", connector.name(), e);
+ return;
+ }
+
+ ConnectorWorker worker;
+ if (isSink(clazz)) {
+ worker = new SinkWorker((Sink) connector, (SinkConfig) config);
+ } else if (isSource(clazz)) {
+ worker = new SourceWorker((Source) connector, (SourceConfig)
config);
+ } else {
+ log.error("class {} is not sink and source", clazz);
+ return;
+ }
+ Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+ worker.stop();
+ log.info("connector {} stopped", connector.name());
+ }));
+ worker.start();
+ log.info("connector {} started", connector.name());
+ }
+
+ private static boolean isAssignableFrom(Class<?> c, Class<?> cls) {
+ Class<?>[] clazzArr = c.getInterfaces();
+ for (Class<?> clazz : clazzArr) {
+ if (clazz.isAssignableFrom(cls)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private static boolean isSink(Class<?> c) {
+ while (c != null && c != Object.class) {
+ if (isAssignableFrom(c, Sink.class)) {
+ return true;
+ }
+ c = c.getSuperclass();
+ }
+ return false;
+ }
+
+ private static boolean isSource(Class<?> c) {
+ while (c != null && c != Object.class) {
+ if (isAssignableFrom(c, Source.class)) {
+ return true;
+ }
+ c = c.getSuperclass();
+ }
+ return false;
+ }
+}
diff --git
a/eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/ConnectorWorker.java
b/eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/ConnectorWorker.java
new file mode 100644
index 000000000..59440077b
--- /dev/null
+++
b/eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/ConnectorWorker.java
@@ -0,0 +1,7 @@
+package org.apache.eventmesh.connector;
+
+public interface ConnectorWorker {
+ void start();
+
+ void stop();
+}
diff --git
a/eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/SinkWorker.java
b/eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/SinkWorker.java
new file mode 100644
index 000000000..3c22e6db2
--- /dev/null
+++
b/eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/SinkWorker.java
@@ -0,0 +1,145 @@
+/*
+ * 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.connector;
+
+import org.apache.eventmesh.client.tcp.EventMeshTCPClient;
+import org.apache.eventmesh.client.tcp.EventMeshTCPClientFactory;
+import org.apache.eventmesh.client.tcp.common.MessageUtils;
+import org.apache.eventmesh.client.tcp.common.ReceiveMsgHook;
+import org.apache.eventmesh.client.tcp.conf.EventMeshTCPClientConfig;
+import org.apache.eventmesh.common.protocol.SubscriptionMode;
+import org.apache.eventmesh.common.protocol.SubscriptionType;
+import org.apache.eventmesh.common.protocol.tcp.UserAgent;
+import org.apache.eventmesh.common.utils.SystemUtils;
+import org.apache.eventmesh.connector.api.config.SinkConfig;
+import org.apache.eventmesh.connector.api.data.ConnectRecord;
+import org.apache.eventmesh.connector.api.sink.Sink;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+
+import io.cloudevents.CloudEvent;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class SinkWorker implements ConnectorWorker {
+
+ private final Sink sink;
+ private final SinkConfig config;
+
+ private final EventMeshTCPClient<CloudEvent> eventMeshTCPClient;
+
+ public SinkWorker(Sink sink, SinkConfig config) throws Exception {
+ this.sink = sink;
+ this.config = config;
+ sink.init(config);
+ eventMeshTCPClient = buildEventMeshSubClient(config);
+ eventMeshTCPClient.init();
+ }
+
+ private EventMeshTCPClient<CloudEvent> buildEventMeshSubClient(SinkConfig
config) {
+ String meshAddress = config.getPubSubConfig().getMeshAddress();
+ String meshIp = meshAddress.split(":")[0];
+ int meshPort = Integer.parseInt(meshAddress.split(":")[1]);
+ UserAgent agent = UserAgent.builder()
+ .env(config.getPubSubConfig().getEnv())
+ .host("localhost")
+ .password(config.getPubSubConfig().getPassWord())
+ .username(config.getPubSubConfig().getUserName())
+ .group(config.getPubSubConfig().getGroup())
+ .path("/")
+ .port(8362)
+ .subsystem(config.getPubSubConfig().getSubsystem())
+ .pid(Integer.parseInt(SystemUtils.getProcessId()))
+ .version("2.0")
+ .idc(config.getPubSubConfig().getIdc())
+ .build();
+ UserAgent userAgent = MessageUtils.generateSubClient(agent);
+
+ EventMeshTCPClientConfig eventMeshTcpClientConfig =
EventMeshTCPClientConfig.builder()
+ .host(meshIp)
+ .port(meshPort)
+ .userAgent(userAgent)
+ .build();
+ return
EventMeshTCPClientFactory.createEventMeshTCPClient(eventMeshTcpClientConfig,
CloudEvent.class);
+ }
+
+ @Override
+ public void start() {
+ log.info("sink worker starting {}", sink.name());
+ log.info("event mesh address is {}",
config.getPubSubConfig().getMeshAddress());
+ try {
+ sink.start();
+ } catch (Exception e) {
+ e.printStackTrace();
+ log.error(e.getMessage());
+ }
+ eventMeshTCPClient.subscribe(config.getPubSubConfig().getMeshTopic(),
SubscriptionMode.CLUSTERING,
+ SubscriptionType.ASYNC);
+ eventMeshTCPClient.registerSubBusiHandler(new EventHandler(sink));
+ eventMeshTCPClient.listen();
+ }
+
+ @Override
+ public void stop() {
+ log.info("sink worker stopping");
+ try {
+ eventMeshTCPClient.unsubscribe();
+ eventMeshTCPClient.close();
+ } catch (Exception e) {
+ e.printStackTrace();
+ log.error("event mesh client close", e);
+ }
+ try {
+ sink.stop();
+ } catch (Exception e) {
+ log.error("sink destroy error", e);
+ }
+
+ log.info("source worker stopped");
+ }
+
+ static class EventHandler implements ReceiveMsgHook<CloudEvent> {
+ private final Sink sink;
+
+ public EventHandler(Sink sink) {
+ this.sink = sink;
+ }
+
+ @Override
+ public Optional<CloudEvent> handle(CloudEvent event) {
+ byte[] body = Objects.requireNonNull(event.getData()).toBytes();
+ //todo: recordPartition & recordOffset
+ ConnectRecord connectRecord = new ConnectRecord(null, null,
System.currentTimeMillis(), body);
+ for (String extensionName : event.getExtensionNames()) {
+ connectRecord.addExtension(extensionName,
Objects.requireNonNull(event.getExtension(extensionName)).toString());
+ }
+ connectRecord.addExtension("id", event.getId());
+ connectRecord.addExtension("topic", event.getSubject());
+ connectRecord.addExtension("source", event.getSource().toString());
+ connectRecord.addExtension("type", event.getType());
+ List<ConnectRecord> connectRecords = new ArrayList<>();
+ connectRecords.add(connectRecord);
+ sink.put(connectRecords);
+ return Optional.empty();
+ }
+ }
+}
diff --git
a/eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/SourceWorker.java
b/eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/SourceWorker.java
new file mode 100644
index 000000000..01fd26bfc
--- /dev/null
+++
b/eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/SourceWorker.java
@@ -0,0 +1,146 @@
+/*
+ * 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.connector;
+
+import org.apache.eventmesh.client.tcp.EventMeshTCPClient;
+import org.apache.eventmesh.client.tcp.EventMeshTCPClientFactory;
+import org.apache.eventmesh.client.tcp.common.EventMeshCommon;
+import org.apache.eventmesh.client.tcp.common.MessageUtils;
+import org.apache.eventmesh.client.tcp.conf.EventMeshTCPClientConfig;
+import org.apache.eventmesh.common.protocol.tcp.UserAgent;
+import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.common.utils.SystemUtils;
+import org.apache.eventmesh.connector.api.config.SourceConfig;
+import org.apache.eventmesh.connector.api.data.ConnectRecord;
+import org.apache.eventmesh.connector.api.source.Source;
+
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.UUID;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class SourceWorker implements ConnectorWorker {
+
+ private final Source source;
+ private final SourceConfig config;
+
+ private final EventMeshTCPClient<CloudEvent> eventMeshTCPClient;
+
+ private volatile boolean isRunning = true;
+
+ public SourceWorker(Source source, SourceConfig config) throws Exception {
+ this.source = source;
+ this.config = config;
+ source.init(config);
+ eventMeshTCPClient = buildEventMeshPubClient(config);
+ eventMeshTCPClient.init();
+ }
+
+ private EventMeshTCPClient<CloudEvent>
buildEventMeshPubClient(SourceConfig config) {
+ String meshAddress = config.getPubSubConfig().getMeshAddress();
+ String meshIp = meshAddress.split(":")[0];
+ int meshPort = Integer.parseInt(meshAddress.split(":")[1]);
+ UserAgent agent = UserAgent.builder()
+ .env(config.getPubSubConfig().getEnv())
+ .host("localhost")
+ .password(config.getPubSubConfig().getPassWord())
+ .username(config.getPubSubConfig().getUserName())
+ .group(config.getPubSubConfig().getGroup())
+ .path("/")
+ .port(8362)
+ .subsystem(config.getPubSubConfig().getSubsystem())
+ .pid(Integer.parseInt(SystemUtils.getProcessId()))
+ .version("2.0")
+ .idc(config.getPubSubConfig().getIdc())
+ .build();
+ UserAgent userAgent = MessageUtils.generatePubClient(agent);
+
+ EventMeshTCPClientConfig eventMeshTcpClientConfig =
EventMeshTCPClientConfig.builder()
+ .host(meshIp)
+ .port(meshPort)
+ .userAgent(userAgent)
+ .build();
+ return
EventMeshTCPClientFactory.createEventMeshTCPClient(eventMeshTcpClientConfig,
CloudEvent.class);
+ }
+
+ @Override
+ public void start() {
+ log.info("source worker starting {}", source.name());
+ log.info("event mesh address is {}",
config.getPubSubConfig().getMeshAddress());
+ try {
+ source.start();
+ while (isRunning) {
+ List<ConnectRecord> connectorRecordList = source.poll();
+ for (ConnectRecord connectRecord : connectorRecordList) {
+ // todo: convert connectRecord to cloudevent
+ CloudEvent event = convertRecordToEvent(connectRecord);
+ eventMeshTCPClient.publish(event, 3000);
+ }
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ log.error(e.getMessage());
+ }
+ log.info("source worker started");
+ }
+
+ private CloudEvent convertRecordToEvent(ConnectRecord connectRecord) {
+ final Map<String, String> content = new HashMap<>();
+ content.put("content", connectRecord.getData().toString());
+
+ return CloudEventBuilder.v1()
+ .withId(UUID.randomUUID().toString())
+ .withSubject(config.getPubSubConfig().getMeshTopic())
+ .withSource(URI.create("/"))
+ .withDataContentType("application/cloudevents+json")
+ .withType(EventMeshCommon.CLOUD_EVENTS_PROTOCOL_NAME)
+
.withData(Objects.requireNonNull(JsonUtils.toJSONString(content)).getBytes(StandardCharsets.UTF_8))
+ .withExtension("ttl", 10000)
+ .build();
+ }
+
+ @Override
+ public void stop() {
+ log.info("source worker stopping");
+ isRunning = false;
+ try {
+ source.stop();
+ } catch (Exception e) {
+ e.printStackTrace();
+ log.error("source destroy error", e);
+ }
+
+ try {
+ eventMeshTCPClient.close();
+ } catch (Exception e) {
+ e.printStackTrace();
+ log.error("event mesh client close", e);
+ }
+ log.info("source worker stopped");
+ }
+
+}
diff --git
a/eventmesh-connectors/eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/api/config/Config.java
b/eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/api/config/Config.java
similarity index 100%
rename from
eventmesh-connectors/eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/api/config/Config.java
rename to
eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/api/config/Config.java
diff --git
a/eventmesh-connectors/eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/api/config/SinkConfig.java
b/eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/api/config/Constants.java
similarity index 73%
copy from
eventmesh-connectors/eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/api/config/SinkConfig.java
copy to
eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/api/config/Constants.java
index 2b28c28f3..204a559c7 100644
---
a/eventmesh-connectors/eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/api/config/SinkConfig.java
+++
b/eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/api/config/Constants.java
@@ -17,5 +17,15 @@
package org.apache.eventmesh.connector.api.config;
-public abstract class SinkConfig extends Config {
+public class Constants {
+
+ public static final String ENV_TARGET = "connectorTarget";
+
+ public static final String ENV_PORT = "connectorPort";
+
+ public static final String ENV_CONFIG_FILE = "connectorConf";
+
+ public static final int DEFAULT_ATTEMPT = 3;
+
+ public static final int DEFAULT_PORT = 8080;
}
diff --git
a/eventmesh-connectors/eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/api/config/SinkConfig.java
b/eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/api/config/PubSubConfig.java
similarity index 75%
copy from
eventmesh-connectors/eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/api/config/SinkConfig.java
copy to
eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/api/config/PubSubConfig.java
index 2b28c28f3..7423de203 100644
---
a/eventmesh-connectors/eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/api/config/SinkConfig.java
+++
b/eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/api/config/PubSubConfig.java
@@ -17,5 +17,25 @@
package org.apache.eventmesh.connector.api.config;
-public abstract class SinkConfig extends Config {
+import lombok.Data;
+
+@Data
+public class PubSubConfig {
+
+ private String meshAddress;
+
+ private String subject;
+
+ private String idc;
+
+ private String env;
+
+ private String group;
+
+ private String appId;
+
+ private String userName;
+
+ private String passWord;
+
}
diff --git
a/eventmesh-connectors/eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/api/config/SinkConfig.java
b/eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/api/config/SinkConfig.java
similarity index 93%
rename from
eventmesh-connectors/eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/api/config/SinkConfig.java
rename to
eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/api/config/SinkConfig.java
index 2b28c28f3..2a4621d7f 100644
---
a/eventmesh-connectors/eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/api/config/SinkConfig.java
+++
b/eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/api/config/SinkConfig.java
@@ -17,5 +17,11 @@
package org.apache.eventmesh.connector.api.config;
+import lombok.Data;
+
+@Data
public abstract class SinkConfig extends Config {
+
+ private PubSubConfig pubSubConfig;
+
}
diff --git
a/eventmesh-connectors/eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/api/config/SourceConfig.java
b/eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/api/config/SourceConfig.java
similarity index 93%
rename from
eventmesh-connectors/eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/api/config/SourceConfig.java
rename to
eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/api/config/SourceConfig.java
index 3aa81f10e..9b74d5e36 100644
---
a/eventmesh-connectors/eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/api/config/SourceConfig.java
+++
b/eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/api/config/SourceConfig.java
@@ -17,5 +17,11 @@
package org.apache.eventmesh.connector.api.config;
+import lombok.Data;
+
+@Data
public abstract class SourceConfig extends Config {
+
+ private PubSubConfig pubSubConfig;
+
}
diff --git
a/eventmesh-connectors/eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/api/connector/Connector.java
b/eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/api/connector/Connector.java
similarity index 100%
rename from
eventmesh-connectors/eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/api/connector/Connector.java
rename to
eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/api/connector/Connector.java
diff --git
a/eventmesh-connectors/eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/api/connector/SinkConnector.java
b/eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/api/connector/SinkConnector.java
similarity index 100%
rename from
eventmesh-connectors/eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/api/connector/SinkConnector.java
rename to
eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/api/connector/SinkConnector.java
diff --git
a/eventmesh-connectors/eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/api/connector/SourceConnector.java
b/eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/api/connector/SourceConnector.java
similarity index 100%
rename from
eventmesh-connectors/eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/api/connector/SourceConnector.java
rename to
eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/api/connector/SourceConnector.java
diff --git
a/eventmesh-connectors/eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/api/data/ConnectRecord.java
b/eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/api/data/ConnectRecord.java
similarity index 100%
rename from
eventmesh-connectors/eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/api/data/ConnectRecord.java
rename to
eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/api/data/ConnectRecord.java
diff --git
a/eventmesh-connectors/eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/api/data/DefaultKeyValue.java
b/eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/api/data/DefaultKeyValue.java
similarity index 100%
rename from
eventmesh-connectors/eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/api/data/DefaultKeyValue.java
rename to
eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/api/data/DefaultKeyValue.java
diff --git
a/eventmesh-connectors/eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/api/data/KeyValue.java
b/eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/api/data/KeyValue.java
similarity index 100%
rename from
eventmesh-connectors/eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/api/data/KeyValue.java
rename to
eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/api/data/KeyValue.java
diff --git
a/eventmesh-connectors/eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/api/data/RecordOffset.java
b/eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/api/data/RecordOffset.java
similarity index 100%
rename from
eventmesh-connectors/eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/api/data/RecordOffset.java
rename to
eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/api/data/RecordOffset.java
diff --git
a/eventmesh-connectors/eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/api/data/RecordPartition.java
b/eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/api/data/RecordPartition.java
similarity index 100%
rename from
eventmesh-connectors/eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/api/data/RecordPartition.java
rename to
eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/api/data/RecordPartition.java
diff --git
a/eventmesh-connectors/eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/api/data/RecordPosition.java
b/eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/api/data/RecordPosition.java
similarity index 100%
rename from
eventmesh-connectors/eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/api/data/RecordPosition.java
rename to
eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/api/data/RecordPosition.java
diff --git
a/eventmesh-connectors/eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/api/sink/Sink.java
b/eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/api/sink/Sink.java
similarity index 100%
rename from
eventmesh-connectors/eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/api/sink/Sink.java
rename to
eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/api/sink/Sink.java
diff --git
a/eventmesh-connectors/eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/api/source/Source.java
b/eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/api/source/Source.java
similarity index 100%
rename from
eventmesh-connectors/eventmesh-connector-api/src/main/java/org/apache/eventmesh/connector/api/source/Source.java
rename to
eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/api/source/Source.java
diff --git
a/eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/util/ConfigUtil.java
b/eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/util/ConfigUtil.java
new file mode 100644
index 000000000..bb1f53008
--- /dev/null
+++
b/eventmesh-connector-sdks/eventmesh-connector-sdk-java/src/main/java/org/apache/eventmesh/connector/util/ConfigUtil.java
@@ -0,0 +1,69 @@
+/*
+ * 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.connector.util;
+
+import org.apache.eventmesh.connector.api.config.Config;
+import org.apache.eventmesh.connector.api.config.Constants;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.net.URL;
+
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class ConfigUtil {
+
+ public static Config parse(Class<? extends Config> c) throws Exception {
+ if (c == null) {
+ return null;
+ }
+ return parseConfig(c);
+ }
+
+ private static Config parseConfig(Class<? extends Config> c) throws
Exception {
+ String configFile = System.getProperty(Constants.ENV_CONFIG_FILE,
System.getenv(Constants.ENV_CONFIG_FILE));
+ if (configFile == null || configFile.isEmpty()) {
+ configFile = "config.yml";
+ }
+ return parse(c, configFile);
+ }
+
+ private static <T> T parse(Class<T> c, String filePathName) throws
Exception {
+ ObjectMapper objectMapper;
+ if (filePathName.endsWith("json")) {
+ objectMapper = new ObjectMapper();
+ } else {
+ objectMapper = new ObjectMapper(new YAMLFactory());
+ }
+
objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES,
false);
+ File file = new File(filePathName);
+ if (file.exists()) {
+ return objectMapper.readValue(file, c);
+ }
+ URL url = ConfigUtil.class.getClassLoader().getResource(filePathName);
+ if (url == null) {
+ throw new FileNotFoundException(filePathName);
+ }
+ return objectMapper.readValue(url, c);
+ }
+}
diff --git a/eventmesh-connectors/eventmesh-connector-api/gradle.properties
b/eventmesh-connector-sdks/gradle.properties
similarity index 100%
rename from eventmesh-connectors/eventmesh-connector-api/gradle.properties
rename to eventmesh-connector-sdks/gradle.properties
diff --git a/settings.gradle b/settings.gradle
index e1e742ee3..866656e05 100644
--- a/settings.gradle
+++ b/settings.gradle
@@ -23,7 +23,8 @@ include 'eventmesh-common'
include 'eventmesh-starter'
include 'eventmesh-examples'
include 'eventmesh-spi'
-include 'eventmesh-connectors:eventmesh-connector-api'
+include 'eventmesh-connector-sdks:eventmesh-connector-sdk-java'
+//include 'eventmesh-connectors:eventmesh-connector-sdk-java'
include 'eventmesh-connectors:source-connector-rocketmq'
include 'eventmesh-connectors:sink-connector-rocketmq'
include 'eventmesh-storage-plugin:eventmesh-storage-api'
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
