add canal connector

eventmesh-common/src/main/java/org/apache/eventmesh/common/config/connector/rdb/canal/CanalSinkConfig.java

@@ -0,0 +1,15 @@
+package org.apache.eventmesh.common.config.connector.rdb.canal;
+
+import org.apache.eventmesh.common.config.connector.SinkConfig;
+import org.apache.eventmesh.common.config.connector.rdb.jdbc.SinkConnectorConfig;
+
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+@Data
+@EqualsAndHashCode(callSuper = true)
+public class CanalSinkConfig extends SinkConfig {
+
+    public SinkConnectorConfig sinkConnectorConfig;
+
+}

eventmesh-common/src/main/java/org/apache/eventmesh/common/config/connector/rdb/canal/CanalSourceConfig.java

@@ -0,0 +1,14 @@
+package org.apache.eventmesh.common.config.connector.rdb.canal;
+
+import org.apache.eventmesh.common.config.connector.SourceConfig;
+import org.apache.eventmesh.common.config.connector.rdb.jdbc.SourceConnectorConfig;
+
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+@Data
+@EqualsAndHashCode(callSuper = true)
+public class CanalSourceConfig extends SourceConfig {
+
+    private SourceConnectorConfig sourceConnectorConfig;
+}

eventmesh-common/src/main/java/org/apache/eventmesh/common/config/connector/rdb/canal/SinkConnectorConfig.java

@@ -0,0 +1,30 @@
+/*
+ * 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.common.config.connector.rdb.canal;
+
+import org.apache.eventmesh.common.config.connector.rdb.jdbc.JdbcConfig;
+
+import lombok.Data;
+
+/**
+ * Configuration parameters for a sink connector.
+ */
+@Data
+public class SinkConnectorConfig {
+
+}

eventmesh-common/src/main/java/org/apache/eventmesh/common/config/connector/rdb/canal/SourceConnectorConfig.java

@@ -0,0 +1,34 @@
+/*
+ * 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.common.config.connector.rdb.canal;
+
+import org.apache.eventmesh.common.config.connector.rdb.jdbc.JdbcConfig;
+import org.apache.eventmesh.common.config.connector.rdb.jdbc.MysqlConfig;
+
+import java.util.List;
+
+import lombok.Data;
+
+/**
+ * Represents the configuration for a database connector.
+ */
+@Data
+public class SourceConnectorConfig {
+
+    private String connectorName;
+}

eventmesh-connectors/eventmesh-connector-canal/build.gradle

@@ -0,0 +1,32 @@
+/*
+ * 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 canal = [
+        "com.alibaba.otter:canal.instance.manager:$canal_version",
+        "com.alibaba.otter:canal.parse:$canal_version",
+        "com.alibaba.otter:canal.server:$canal_version"
+]
+
+dependencies {
+    api project(":eventmesh-openconnect:eventmesh-openconnect-java")
+    implementation project(":eventmesh-common")
+    implementation canal
+    compileOnly 'org.projectlombok:lombok'
+    annotationProcessor 'org.projectlombok:lombok'
+    testImplementation "org.mockito:mockito-core"
+    testImplementation "org.mockito:mockito-junit-jupiter"
+}

eventmesh-connectors/eventmesh-connector-canal/gradle.properties

@@ -0,0 +1,19 @@
+#
+# 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.
+#
+canal_version=1.1.7
+pluginType=connector
+pluginName=canal

eventmesh-connectors/eventmesh-connector-canal/src/main/java/org/apache/eventmesh/connector/canal/config/CanalServerConfig.java

@@ -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.
+ */
+
+package org.apache.eventmesh.connector.canal.config;
+
+import org.apache.eventmesh.common.config.connector.Config;
+
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+@Data
+@EqualsAndHashCode(callSuper = true)
+public class CanalServerConfig extends Config {
+
+    private boolean sourceEnable;
+
+    private boolean sinkEnable;
+
+}

eventmesh-connectors/eventmesh-connector-canal/src/main/java/org/apache/eventmesh/connector/canal/server/CanalConnectServer.java

@@ -0,0 +1,45 @@
+/*
+ * 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.canal.server;
+
+import org.apache.eventmesh.connector.canal.config.CanalServerConfig;
+import org.apache.eventmesh.connector.canal.sink.connector.CanalSinkConnector;
+import org.apache.eventmesh.connector.canal.source.connector.CanalSourceConnector;
+import org.apache.eventmesh.openconnect.Application;
+import org.apache.eventmesh.openconnect.util.ConfigUtil;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class CanalConnectServer {
+
+    public static void main(String[] args) throws Exception {
+
+        CanalServerConfig serverConfig = ConfigUtil.parse(CanalServerConfig.class, "server-config.yml");
+
+        if (serverConfig.isSourceEnable()) {
+            Application canalSourceApp = new Application();
+            canalSourceApp.run(CanalSourceConnector.class);
+        }
+
+        if (serverConfig.isSinkEnable()) {
+            Application canalSinkApp = new Application();
+            canalSinkApp.run(CanalSinkConnector.class);
+        }
+    }
+}

eventmesh-connectors/eventmesh-connector-canal/src/main/java/org/apache/eventmesh/connector/canal/sink/connector/CanalSinkConnector.java

@@ -0,0 +1,84 @@
+/*
+ * 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.canal.sink.connector;
+
+import org.apache.eventmesh.common.config.connector.Config;
+
+import org.apache.eventmesh.common.config.connector.rdb.canal.CanalSinkConfig;
+import org.apache.eventmesh.openconnect.api.ConnectorCreateService;
+import org.apache.eventmesh.openconnect.api.connector.ConnectorContext;
+import org.apache.eventmesh.openconnect.api.connector.SinkConnectorContext;
+import org.apache.eventmesh.openconnect.api.sink.Sink;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.ConnectRecord;
+
+
+import java.util.List;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class CanalSinkConnector implements Sink, ConnectorCreateService<Sink> {
+
+    @Override
+    public Class<? extends Config> configClass() {
+        return CanalSinkConfig.class;
+    }
+
+    @Override
+    public void init(Config config) throws Exception {
+        // init config for canal source connector
+
+    }
+
+    @Override
+    public void init(ConnectorContext connectorContext) throws Exception {
+        // init config for canal source connector
+        SinkConnectorContext sinkConnectorContext = (SinkConnectorContext) connectorContext;
+
+    }
+
+    @Override
+    public void start() throws Exception {
+
+    }
+
+    @Override
+    public void commit(ConnectRecord record) {
+
+    }
+
+    @Override
+    public String name() {
+        return "";
+    }
+
+    @Override
+    public void stop() {
+
+    }
+
+    @Override
+    public void put(List<ConnectRecord> sinkRecords) {
+
+    }
+
+    @Override
+    public Sink create() {
+        return new CanalSinkConnector();
+    }
+}