This is an automated email from the ASF dual-hosted git repository.
mxsm 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 d327ff3ed [ISSUE #4413] Add spring boot source connector. (#4477)
d327ff3ed is described below
commit d327ff3edbfc7cddac6458d84a33f6db55cc16ad
Author: yanrongzhen <[email protected]>
AuthorDate: Sat Oct 14 13:32:06 2023 +0800
[ISSUE #4413] Add spring boot source connector. (#4477)
* Add spring boot source connector.
* fix license check.
* Implement the Spring source connector.
* Delete unused files.
* Fix interface naming
* fix function naming.
* Revert sdk changes.
---
.../org/apache/eventmesh/common/Constants.java | 4 +
.../common/protocol/grpc/common/Response.java | 1 -
.../eventmesh-connector-spring/build.gradle | 37 +++----
.../eventmesh-connector-spring/gradle.properties | 18 ++++
.../common/SpringApplicationContextHolder.java | 44 ++++++++
.../spring/config/EventMeshAutoConfiguration.java | 55 ++++++++++
.../spring/config/SpringConnectServerConfig.java | 22 ++--
.../spring/server/SpringConnectServer.java | 40 +++++++
.../spring/source/MessageSendingOperations.java | 27 ++---
.../spring/source/config/SpringSourceConfig.java | 23 ++---
.../source/connector/SourceConnectorConfig.java | 21 +---
.../source/connector/SpringSourceConnector.java | 115 +++++++++++++++++++++
.../SpringSourceConnectorCreateServiceImpl.java | 40 +++----
...mesh.openconnect.api.source.SourceCreateService | 18 ++++
.../src/main/resources/META-INF/spring.factories | 19 ++++
.../src/main/resources/server-config.yml | 19 ++++
.../src/main/resources/source-config.yml | 29 ++++++
.../apache/eventmesh/openconnect/Application.java | 18 +++-
.../apache/eventmesh/openconnect/SourceWorker.java | 10 +-
.../api/source/SourceCreateService.java | 30 ++----
.../DefaultOffsetManagementServiceImpl.java | 82 +++++++++++++++
settings.gradle | 2 +
22 files changed, 525 insertions(+), 149 deletions(-)
diff --git
a/eventmesh-common/src/main/java/org/apache/eventmesh/common/Constants.java
b/eventmesh-common/src/main/java/org/apache/eventmesh/common/Constants.java
index d9f50ab47..9a1dc0686 100644
--- a/eventmesh-common/src/main/java/org/apache/eventmesh/common/Constants.java
+++ b/eventmesh-common/src/main/java/org/apache/eventmesh/common/Constants.java
@@ -32,6 +32,8 @@ public class Constants {
public static final String LANGUAGE_JAVA = "JAVA";
+ public static final String CONNECT_SERVER_CONFIG_FILE_NAME =
"server-config.yml";
+
public static final String HTTP_PROTOCOL_PREFIX = "http://";;
public static final String HTTPS_PROTOCOL_PREFIX = "https://";;
@@ -149,6 +151,8 @@ public class Constants {
public static final String UNDER_LINE = "_";
+ public static final String UNKNOWN = "unknown";
+
public static final String LEFT_PARENTHESIS = "(";
public static final String RIGHT_PARENTHESIS = ")";
diff --git
a/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/common/Response.java
b/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/common/Response.java
index d175dce47..965a45437 100644
---
a/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/common/Response.java
+++
b/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/common/Response.java
@@ -17,7 +17,6 @@
package org.apache.eventmesh.common.protocol.grpc.common;
-
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
diff --git
a/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/common/Response.java
b/eventmesh-connectors/eventmesh-connector-spring/build.gradle
similarity index 55%
copy from
eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/common/Response.java
copy to eventmesh-connectors/eventmesh-connector-spring/build.gradle
index d175dce47..06057096a 100644
---
a/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/common/Response.java
+++ b/eventmesh-connectors/eventmesh-connector-spring/build.gradle
@@ -15,28 +15,17 @@
* limitations under the License.
*/
-package org.apache.eventmesh.common.protocol.grpc.common;
-
-
-import lombok.AllArgsConstructor;
-import lombok.Builder;
-import lombok.Data;
-import lombok.EqualsAndHashCode;
-import lombok.NoArgsConstructor;
-import lombok.ToString;
-
-@Data
-@AllArgsConstructor
-@NoArgsConstructor
-@EqualsAndHashCode
-@Builder
-@ToString
-public class Response {
-
- private String respCode;
-
- private String respMsg;
-
- private String respTime;
-
+configurations {
+ implementation.exclude group: 'ch.qos.logback', module: 'logback-classic'
+ implementation.exclude group: 'log4j', module: 'log4j'
}
+
+dependencies {
+ api project(":eventmesh-openconnect:eventmesh-openconnect-java")
+ implementation project(":eventmesh-sdks:eventmesh-sdk-java")
+ implementation
"org.springframework.boot:spring-boot-starter:$spring_boot_version"
+ implementation
"org.springframework.boot:spring-boot-starter-validation:$spring_boot_version"
+ implementation "org.springframework:spring-messaging:$spring_version"
+ compileOnly 'org.projectlombok:lombok'
+ annotationProcessor 'org.projectlombok:lombok'
+}
\ No newline at end of file
diff --git a/eventmesh-connectors/eventmesh-connector-spring/gradle.properties
b/eventmesh-connectors/eventmesh-connector-spring/gradle.properties
new file mode 100644
index 000000000..37f076c98
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-spring/gradle.properties
@@ -0,0 +1,18 @@
+#
+# 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.
+#
+spring_boot_version=2.5.9
+spring_version=5.3.20
\ No newline at end of file
diff --git
a/eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/common/SpringApplicationContextHolder.java
b/eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/common/SpringApplicationContextHolder.java
new file mode 100644
index 000000000..aa110b2aa
--- /dev/null
+++
b/eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/common/SpringApplicationContextHolder.java
@@ -0,0 +1,44 @@
+/*
+ * 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.spring.common;
+
+import org.springframework.beans.BeansException;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationContextAware;
+
+public class SpringApplicationContextHolder implements ApplicationContextAware
{
+
+ private static ApplicationContext applicationContext;
+
+ @Override
+ public void setApplicationContext(ApplicationContext applicationContext)
throws BeansException {
+ SpringApplicationContextHolder.applicationContext = applicationContext;
+ }
+
+ public static Object getBean(String beanName) {
+ return applicationContext.getBean(beanName);
+ }
+
+ public static Object getBean(Class<?> beanClass) {
+ return applicationContext.getBean(beanClass);
+ }
+
+ public static boolean isStarted() {
+ return applicationContext != null;
+ }
+}
diff --git
a/eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/config/EventMeshAutoConfiguration.java
b/eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/config/EventMeshAutoConfiguration.java
new file mode 100644
index 000000000..008efef8f
--- /dev/null
+++
b/eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/config/EventMeshAutoConfiguration.java
@@ -0,0 +1,55 @@
+/*
+ * 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.spring.config;
+
+import
org.apache.eventmesh.connector.spring.common.SpringApplicationContextHolder;
+import org.apache.eventmesh.connector.spring.server.SpringConnectServer;
+import
org.apache.eventmesh.connector.spring.source.connector.SpringSourceConnector;
+
+import
org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * Spring auto configuration.
+ */
+@Configuration
+public class EventMeshAutoConfiguration {
+
+ public static final String SPRING_SOURCE_CONNECTOR_BEAN_NAME =
"springSourceConnector";
+ public static final String SPRING_CONNECT_SERVER_BEAN_NAME =
"springConnectServer";
+ public static final String SPRING_APPLICATION_CONTEXT_HOLDER =
"springApplicationContextHolder";
+
+ @Bean(name = SPRING_SOURCE_CONNECTOR_BEAN_NAME)
+ @ConditionalOnMissingBean(name = SPRING_SOURCE_CONNECTOR_BEAN_NAME)
+ public SpringSourceConnector springSourceConnector() {
+ return new SpringSourceConnector();
+ }
+
+ @Bean(name = SPRING_CONNECT_SERVER_BEAN_NAME)
+ @ConditionalOnMissingBean(name = SPRING_CONNECT_SERVER_BEAN_NAME)
+ public SpringConnectServer springConnectServer() {
+ return new SpringConnectServer();
+ }
+
+ @Bean(name = SPRING_APPLICATION_CONTEXT_HOLDER)
+ @ConditionalOnMissingBean(name = SPRING_APPLICATION_CONTEXT_HOLDER)
+ public SpringApplicationContextHolder springApplicationContextHolder() {
+ return new SpringApplicationContextHolder();
+ }
+}
diff --git
a/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/common/Response.java
b/eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/config/SpringConnectServerConfig.java
similarity index 70%
copy from
eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/common/Response.java
copy to
eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/config/SpringConnectServerConfig.java
index d175dce47..7eb2009fb 100644
---
a/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/common/Response.java
+++
b/eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/config/SpringConnectServerConfig.java
@@ -15,28 +15,18 @@
* limitations under the License.
*/
-package org.apache.eventmesh.common.protocol.grpc.common;
+package org.apache.eventmesh.connector.spring.config;
+import org.apache.eventmesh.openconnect.api.config.Config;
-import lombok.AllArgsConstructor;
-import lombok.Builder;
import lombok.Data;
import lombok.EqualsAndHashCode;
-import lombok.NoArgsConstructor;
-import lombok.ToString;
@Data
-@AllArgsConstructor
-@NoArgsConstructor
-@EqualsAndHashCode
-@Builder
-@ToString
-public class Response {
+@EqualsAndHashCode(callSuper = true)
+public class SpringConnectServerConfig extends Config {
- private String respCode;
-
- private String respMsg;
-
- private String respTime;
+ private boolean sourceEnable;
+ private boolean sinkEnable;
}
diff --git
a/eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/server/SpringConnectServer.java
b/eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/server/SpringConnectServer.java
new file mode 100644
index 000000000..2c79b4b46
--- /dev/null
+++
b/eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/server/SpringConnectServer.java
@@ -0,0 +1,40 @@
+/*
+ * 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.spring.server;
+
+import org.apache.eventmesh.common.Constants;
+import org.apache.eventmesh.connector.spring.config.SpringConnectServerConfig;
+import
org.apache.eventmesh.connector.spring.source.connector.SpringSourceConnector;
+import org.apache.eventmesh.openconnect.Application;
+import org.apache.eventmesh.openconnect.util.ConfigUtil;
+
+import org.springframework.boot.CommandLineRunner;
+
+public class SpringConnectServer implements CommandLineRunner {
+
+ @Override
+ public void run(String... args) throws Exception {
+ SpringConnectServerConfig springConnectServerConfig =
ConfigUtil.parse(SpringConnectServerConfig.class,
+ Constants.CONNECT_SERVER_CONFIG_FILE_NAME);
+
+ if (springConnectServerConfig.isSourceEnable()) {
+ Application application = new Application();
+ application.run(SpringSourceConnector.class);
+ }
+ }
+}
diff --git
a/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/common/Response.java
b/eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/source/MessageSendingOperations.java
similarity index 65%
copy from
eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/common/Response.java
copy to
eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/source/MessageSendingOperations.java
index d175dce47..37c0b291b 100644
---
a/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/common/Response.java
+++
b/eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/source/MessageSendingOperations.java
@@ -15,28 +15,13 @@
* limitations under the License.
*/
-package org.apache.eventmesh.common.protocol.grpc.common;
+package org.apache.eventmesh.connector.spring.source;
+/**
+ * Operations for sending messages.
+ */
+public interface MessageSendingOperations {
-import lombok.AllArgsConstructor;
-import lombok.Builder;
-import lombok.Data;
-import lombok.EqualsAndHashCode;
-import lombok.NoArgsConstructor;
-import lombok.ToString;
-
-@Data
-@AllArgsConstructor
-@NoArgsConstructor
-@EqualsAndHashCode
-@Builder
-@ToString
-public class Response {
-
- private String respCode;
-
- private String respMsg;
-
- private String respTime;
+ void send(Object message);
}
diff --git
a/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/common/Response.java
b/eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/source/config/SpringSourceConfig.java
similarity index 70%
copy from
eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/common/Response.java
copy to
eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/source/config/SpringSourceConfig.java
index d175dce47..191e9b119 100644
---
a/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/common/Response.java
+++
b/eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/source/config/SpringSourceConfig.java
@@ -15,28 +15,17 @@
* limitations under the License.
*/
-package org.apache.eventmesh.common.protocol.grpc.common;
+package org.apache.eventmesh.connector.spring.source.config;
+import
org.apache.eventmesh.connector.spring.source.connector.SourceConnectorConfig;
+import org.apache.eventmesh.openconnect.api.config.SourceConfig;
-import lombok.AllArgsConstructor;
-import lombok.Builder;
import lombok.Data;
import lombok.EqualsAndHashCode;
-import lombok.NoArgsConstructor;
-import lombok.ToString;
@Data
-@AllArgsConstructor
-@NoArgsConstructor
-@EqualsAndHashCode
-@Builder
-@ToString
-public class Response {
-
- private String respCode;
-
- private String respMsg;
-
- private String respTime;
+@EqualsAndHashCode(callSuper = true)
+public class SpringSourceConfig extends SourceConfig {
+ private SourceConnectorConfig sourceConnectorConfig;
}
diff --git
a/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/common/Response.java
b/eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/source/connector/SourceConnectorConfig.java
similarity index 67%
copy from
eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/common/Response.java
copy to
eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/source/connector/SourceConnectorConfig.java
index d175dce47..242869087 100644
---
a/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/common/Response.java
+++
b/eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/source/connector/SourceConnectorConfig.java
@@ -15,28 +15,13 @@
* limitations under the License.
*/
-package org.apache.eventmesh.common.protocol.grpc.common;
+package org.apache.eventmesh.connector.spring.source.connector;
-
-import lombok.AllArgsConstructor;
-import lombok.Builder;
import lombok.Data;
-import lombok.EqualsAndHashCode;
-import lombok.NoArgsConstructor;
-import lombok.ToString;
@Data
-@AllArgsConstructor
-@NoArgsConstructor
-@EqualsAndHashCode
-@Builder
-@ToString
-public class Response {
-
- private String respCode;
-
- private String respMsg;
+public class SourceConnectorConfig {
- private String respTime;
+ private String connectorName;
}
diff --git
a/eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/source/connector/SpringSourceConnector.java
b/eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/source/connector/SpringSourceConnector.java
new file mode 100644
index 000000000..f87dc42e4
--- /dev/null
+++
b/eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/source/connector/SpringSourceConnector.java
@@ -0,0 +1,115 @@
+/*
+ * 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.spring.source.connector;
+
+import org.apache.eventmesh.connector.spring.source.MessageSendingOperations;
+import org.apache.eventmesh.connector.spring.source.config.SpringSourceConfig;
+import org.apache.eventmesh.openconnect.api.config.Config;
+import org.apache.eventmesh.openconnect.api.connector.ConnectorContext;
+import org.apache.eventmesh.openconnect.api.connector.SourceConnectorContext;
+import org.apache.eventmesh.openconnect.api.source.Source;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.ConnectRecord;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.RecordOffset;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.RecordPartition;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class SpringSourceConnector implements Source, MessageSendingOperations
{
+
+ private static final int DEFAULT_BATCH_SIZE = 10;
+
+ private SpringSourceConfig sourceConfig;
+
+ private BlockingQueue<ConnectRecord> queue;
+
+ @Override
+ public Class<? extends Config> configClass() {
+ return SpringSourceConfig.class;
+ }
+
+ @Override
+ public void init(Config config) throws Exception {
+ // init config for spring source connector
+ this.sourceConfig = (SpringSourceConfig) config;
+ this.queue = new LinkedBlockingQueue<>(1000);
+ }
+
+ @Override
+ public void init(ConnectorContext connectorContext) throws Exception {
+ SourceConnectorContext sourceConnectorContext =
(SourceConnectorContext) connectorContext;
+ // init config for spring source connector
+ this.sourceConfig = (SpringSourceConfig)
sourceConnectorContext.getSourceConfig();
+ this.queue = new LinkedBlockingQueue<>(1000);
+ }
+
+ @Override
+ public void start() throws Exception {
+
+ }
+
+ @Override
+ public void commit(ConnectRecord record) {
+
+ }
+
+ @Override
+ public String name() {
+ return this.sourceConfig.getSourceConnectorConfig().getConnectorName();
+ }
+
+ @Override
+ public void stop() throws Exception {
+
+ }
+
+ @Override
+ public List<ConnectRecord> poll() {
+ List<ConnectRecord> connectRecords = new
ArrayList<>(DEFAULT_BATCH_SIZE);
+
+ for (int count = 0; count < DEFAULT_BATCH_SIZE; ++count) {
+ try {
+ ConnectRecord connectRecord = queue.poll(3, TimeUnit.SECONDS);
+ if (connectRecord == null) {
+ break;
+ }
+ connectRecords.add(connectRecord);
+ } catch (InterruptedException e) {
+ Thread currentThread = Thread.currentThread();
+ log.warn("[SpringSourceConnector] Interrupting thread {} due
to exception {}",
+ currentThread.getName(), e.getMessage());
+ currentThread.interrupt();
+ }
+ }
+ return connectRecords;
+ }
+
+ @Override
+ public void send(Object message) {
+ RecordPartition partition = new RecordPartition();
+ RecordOffset offset = new RecordOffset();
+ ConnectRecord record = new ConnectRecord(partition, offset,
System.currentTimeMillis(), message);
+ queue.offer(record);
+ }
+}
diff --git
a/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/common/Response.java
b/eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/source/connector/SpringSourceConnectorCreateServiceImpl.java
similarity index 53%
copy from
eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/common/Response.java
copy to
eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/source/connector/SpringSourceConnectorCreateServiceImpl.java
index d175dce47..0f61caebc 100644
---
a/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/common/Response.java
+++
b/eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/source/connector/SpringSourceConnectorCreateServiceImpl.java
@@ -15,28 +15,20 @@
* limitations under the License.
*/
-package org.apache.eventmesh.common.protocol.grpc.common;
-
-
-import lombok.AllArgsConstructor;
-import lombok.Builder;
-import lombok.Data;
-import lombok.EqualsAndHashCode;
-import lombok.NoArgsConstructor;
-import lombok.ToString;
-
-@Data
-@AllArgsConstructor
-@NoArgsConstructor
-@EqualsAndHashCode
-@Builder
-@ToString
-public class Response {
-
- private String respCode;
-
- private String respMsg;
-
- private String respTime;
-
+package org.apache.eventmesh.connector.spring.source.connector;
+
+import
org.apache.eventmesh.connector.spring.common.SpringApplicationContextHolder;
+import org.apache.eventmesh.connector.spring.config.EventMeshAutoConfiguration;
+import org.apache.eventmesh.openconnect.api.source.Source;
+import org.apache.eventmesh.openconnect.api.source.SourceCreateService;
+
+public class SpringSourceConnectorCreateServiceImpl implements
SourceCreateService {
+
+ @Override
+ public Source create() {
+ if (SpringApplicationContextHolder.isStarted()) {
+ return (Source)
SpringApplicationContextHolder.getBean(EventMeshAutoConfiguration.SPRING_SOURCE_CONNECTOR_BEAN_NAME);
+ }
+ return null;
+ }
}
diff --git
a/eventmesh-connectors/eventmesh-connector-spring/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.openconnect.api.source.SourceCreateService
b/eventmesh-connectors/eventmesh-connector-spring/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.openconnect.api.source.SourceCreateService
new file mode 100644
index 000000000..b79216363
--- /dev/null
+++
b/eventmesh-connectors/eventmesh-connector-spring/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.openconnect.api.source.SourceCreateService
@@ -0,0 +1,18 @@
+#
+# 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.
+#
+
+spring=org.apache.eventmesh.connector.spring.source.connector.SpringSourceConnectorCreateServiceImpl
\ No newline at end of file
diff --git
a/eventmesh-connectors/eventmesh-connector-spring/src/main/resources/META-INF/spring.factories
b/eventmesh-connectors/eventmesh-connector-spring/src/main/resources/META-INF/spring.factories
new file mode 100644
index 000000000..8cc07dc97
--- /dev/null
+++
b/eventmesh-connectors/eventmesh-connector-spring/src/main/resources/META-INF/spring.factories
@@ -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.
+#
+
+org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
+org.apache.eventmesh.connector.spring.config.EventMeshAutoConfiguration
\ No newline at end of file
diff --git
a/eventmesh-connectors/eventmesh-connector-spring/src/main/resources/server-config.yml
b/eventmesh-connectors/eventmesh-connector-spring/src/main/resources/server-config.yml
new file mode 100644
index 000000000..5f66dd0f6
--- /dev/null
+++
b/eventmesh-connectors/eventmesh-connector-spring/src/main/resources/server-config.yml
@@ -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.
+#
+
+sourceEnable: true
+sinkEnable: true
diff --git
a/eventmesh-connectors/eventmesh-connector-spring/src/main/resources/source-config.yml
b/eventmesh-connectors/eventmesh-connector-spring/src/main/resources/source-config.yml
new file mode 100644
index 000000000..a0de4f218
--- /dev/null
+++
b/eventmesh-connectors/eventmesh-connector-spring/src/main/resources/source-config.yml
@@ -0,0 +1,29 @@
+#
+# 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.
+#
+
+pubSubConfig:
+ meshAddress: 127.0.0.1:10000
+ subject: TEST-TOPIC-SPRING
+ idc: FT
+ env: PRD
+ group: springSource
+ appId: 5033
+ userName: springSourceUser
+ passWord: springPassWord
+connectorConfig:
+ connectorName: springSource
+ topic: SourceTopic
diff --git
a/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/Application.java
b/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/Application.java
index 8a0341f41..740bd1226 100644
---
a/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/Application.java
+++
b/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/Application.java
@@ -23,7 +23,9 @@ import
org.apache.eventmesh.openconnect.api.config.SourceConfig;
import org.apache.eventmesh.openconnect.api.connector.Connector;
import org.apache.eventmesh.openconnect.api.sink.Sink;
import org.apache.eventmesh.openconnect.api.source.Source;
+import org.apache.eventmesh.openconnect.api.source.SourceCreateService;
import org.apache.eventmesh.openconnect.util.ConfigUtil;
+import org.apache.eventmesh.spi.EventMeshExtensionFactory;
import java.util.HashMap;
import java.util.Map;
@@ -35,11 +37,20 @@ public class Application {
public static final Map<String, Connector> CONNECTOR_MAP = new HashMap<>();
+ private static final String SPRING = "spring";
+
public void run(Class<? extends Connector> clazz) throws Exception {
- Connector connector;
+ Connector connector = null;
try {
- connector = clazz.getDeclaredConstructor().newInstance();
+ SourceCreateService createService =
+
EventMeshExtensionFactory.getExtension(SourceCreateService.class, SPRING);
+ if (createService != null) {
+ connector = createService.create();
+ }
+ if (connector == null) {
+ connector = clazz.getDeclaredConstructor().newInstance();
+ }
} catch (Exception e) {
log.error("new connector error", e);
return;
@@ -64,9 +75,10 @@ public class Application {
worker.init();
CONNECTOR_MAP.putIfAbsent(connector.name(), connector);
+ Connector finalConnector = connector;
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
worker.stop();
- log.info("connector {} stopped", connector.name());
+ log.info("connector {} stopped", finalConnector.name());
}));
worker.start();
log.info("connector {} started", connector.name());
diff --git
a/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/SourceWorker.java
b/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/SourceWorker.java
index 009e6e46f..12d3a62f4 100644
---
a/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/SourceWorker.java
+++
b/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/SourceWorker.java
@@ -32,6 +32,7 @@ import org.apache.eventmesh.openconnect.api.source.Source;
import
org.apache.eventmesh.openconnect.offsetmgmt.api.config.OffsetStorageConfig;
import org.apache.eventmesh.openconnect.offsetmgmt.api.data.ConnectRecord;
import
org.apache.eventmesh.openconnect.offsetmgmt.api.data.RecordOffsetManagement;
+import
org.apache.eventmesh.openconnect.offsetmgmt.api.storage.DefaultOffsetManagementServiceImpl;
import
org.apache.eventmesh.openconnect.offsetmgmt.api.storage.OffsetManagementService;
import
org.apache.eventmesh.openconnect.offsetmgmt.api.storage.OffsetStorageReaderImpl;
import
org.apache.eventmesh.openconnect.offsetmgmt.api.storage.OffsetStorageWriterImpl;
@@ -132,12 +133,13 @@ public class SourceWorker implements ConnectorWorker {
}
eventMeshTCPClient.init();
// spi load offsetMgmtService
- OffsetStorageConfig offsetStorageConfig =
config.getOffsetStorageConfig();
- String offsetMgmtPluginType =
offsetStorageConfig.getOffsetStorageType();
this.offsetManagement = new RecordOffsetManagement();
this.committableOffsets =
RecordOffsetManagement.CommittableOffsets.EMPTY;
- this.offsetManagementService =
-
EventMeshExtensionFactory.getExtension(OffsetManagementService.class,
offsetMgmtPluginType);
+ OffsetStorageConfig offsetStorageConfig =
config.getOffsetStorageConfig();
+ this.offsetManagementService = Optional.ofNullable(offsetStorageConfig)
+ .map(OffsetStorageConfig::getOffsetStorageType)
+ .map(storageType ->
EventMeshExtensionFactory.getExtension(OffsetManagementService.class,
storageType))
+ .orElse(new DefaultOffsetManagementServiceImpl());
this.offsetManagementService.initialize(offsetStorageConfig);
this.offsetStorageWriter = new OffsetStorageWriterImpl(source.name(),
offsetManagementService);
this.offsetStorageReader = new OffsetStorageReaderImpl(source.name(),
offsetManagementService);
diff --git
a/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/common/Response.java
b/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/api/source/SourceCreateService.java
similarity index 65%
copy from
eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/common/Response.java
copy to
eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/api/source/SourceCreateService.java
index d175dce47..8ffc0bb5d 100644
---
a/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/common/Response.java
+++
b/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/api/source/SourceCreateService.java
@@ -15,28 +15,16 @@
* limitations under the License.
*/
-package org.apache.eventmesh.common.protocol.grpc.common;
+package org.apache.eventmesh.openconnect.api.source;
+import org.apache.eventmesh.spi.EventMeshExtensionType;
+import org.apache.eventmesh.spi.EventMeshSPI;
-import lombok.AllArgsConstructor;
-import lombok.Builder;
-import lombok.Data;
-import lombok.EqualsAndHashCode;
-import lombok.NoArgsConstructor;
-import lombok.ToString;
-
-@Data
-@AllArgsConstructor
-@NoArgsConstructor
-@EqualsAndHashCode
-@Builder
-@ToString
-public class Response {
-
- private String respCode;
-
- private String respMsg;
-
- private String respTime;
+/**
+ * SPI interface for source connector creation.
+ */
+@EventMeshSPI(eventMeshExtensionType = EventMeshExtensionType.CONNECTOR)
+public interface SourceCreateService {
+ Source create();
}
diff --git
a/eventmesh-openconnect/eventmesh-openconnect-offsetmgmt-plugin/eventmesh-openconnect-offsetmgmt-api/src/main/java/org/apache/eventmesh/openconnect/offsetmgmt/api/storage/DefaultOffsetManagementServiceImpl.java
b/eventmesh-openconnect/eventmesh-openconnect-offsetmgmt-plugin/eventmesh-openconnect-offsetmgmt-api/src/main/java/org/apache/eventmesh/openconnect/offsetmgmt/api/storage/DefaultOffsetManagementServiceImpl.java
new file mode 100644
index 000000000..e31fc358f
--- /dev/null
+++
b/eventmesh-openconnect/eventmesh-openconnect-offsetmgmt-plugin/eventmesh-openconnect-offsetmgmt-api/src/main/java/org/apache/eventmesh/openconnect/offsetmgmt/api/storage/DefaultOffsetManagementServiceImpl.java
@@ -0,0 +1,82 @@
+/*
+ * 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.openconnect.offsetmgmt.api.storage;
+
+import
org.apache.eventmesh.openconnect.offsetmgmt.api.config.OffsetStorageConfig;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.RecordOffset;
+
+import java.util.List;
+import java.util.Map;
+
+public class DefaultOffsetManagementServiceImpl implements
OffsetManagementService {
+
+ @Override
+ public void start() {
+
+ }
+
+ @Override
+ public void stop() {
+
+ }
+
+ @Override
+ public void persist() {
+
+ }
+
+ @Override
+ public void load() {
+
+ }
+
+ @Override
+ public void synchronize() {
+
+ }
+
+ @Override
+ public Map<ConnectorRecordPartition, RecordOffset> getPositionMap() {
+ return null;
+ }
+
+ @Override
+ public RecordOffset getPosition(ConnectorRecordPartition partition) {
+ return null;
+ }
+
+ @Override
+ public void putPosition(Map<ConnectorRecordPartition, RecordOffset>
positions) {
+
+ }
+
+ @Override
+ public void putPosition(ConnectorRecordPartition partition, RecordOffset
position) {
+
+ }
+
+ @Override
+ public void removePosition(List<ConnectorRecordPartition> partitions) {
+
+ }
+
+ @Override
+ public void initialize(OffsetStorageConfig connectorConfig) {
+
+ }
+}
diff --git a/settings.gradle b/settings.gradle
index bdb43facf..5f1b1ed05 100644
--- a/settings.gradle
+++ b/settings.gradle
@@ -40,6 +40,7 @@ include 'eventmesh-connectors:eventmesh-connector-pravega'
include 'eventmesh-connectors:eventmesh-connector-knative'
include 'eventmesh-connectors:eventmesh-connector-jdbc'
include 'eventmesh-connectors:eventmesh-connector-file'
+include 'eventmesh-connectors:eventmesh-connector-spring'
include 'eventmesh-storage-plugin:eventmesh-storage-api'
include 'eventmesh-storage-plugin:eventmesh-storage-standalone'
@@ -88,3 +89,4 @@ include 'eventmesh-webhook'
include 'eventmesh-webhook:eventmesh-webhook-api'
include 'eventmesh-webhook:eventmesh-webhook-admin'
include 'eventmesh-webhook:eventmesh-webhook-receive'
+
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
