[ISSUE #3975] Add eventmesh-connector-sdks module

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"

eventmesh-connectors/eventmesh-connector-api/build.gradle → 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.
- */
+ */

eventmesh-connector-sdks/eventmesh-connector-sdk-java/build.gradle

@@ -0,0 +1,33 @@
+/*
+ * 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.
+ */
+
+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'
+}

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;
+    }
+}

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();
+}

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();
+        }
+    }
+}