This is an automated email from the ASF dual-hosted git repository.
mikexue 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 f6297308f [ISSUE #4414] Add spring sink connector. (#4491)
f6297308f is described below
commit f6297308fc9df28fed22e186a2ac1e6b45b1e712
Author: yanrongzhen <[email protected]>
AuthorDate: Tue Oct 17 16:53:11 2023 +0800
[ISSUE #4414] Add spring sink connector. (#4491)
* Add spring sink connector.
* Restructuring consumption logic.
* Optimize create extension logic.
* Fix naming
---
.../org/apache/eventmesh/common/Constants.java | 4 +
.../apache/eventmesh/common/utils/JsonUtils.java | 19 +++
.../common/SpringApplicationContextHolder.java | 15 ++-
.../spring/config/EventMeshAutoConfiguration.java | 17 +++
.../spring/server/SpringConnectServer.java | 12 +-
.../spring/sink/EventMeshConsumerMetadata.java | 23 ++--
.../connector/spring/sink/EventMeshListener.java | 26 ++--
.../sink/EventMeshListenerBeanPostProcessor.java | 143 +++++++++++++++++++++
.../spring/sink/config/SpringSinkConfig.java | 19 +--
.../spring/sink/connector/SinkConnectorConfig.java | 15 +--
.../spring/sink/connector/SpringSinkConnector.java | 106 +++++++++++++++
.../SpringSinkConnectorCreateServiceImpl.java} | 12 +-
.../SpringSourceConnectorCreateServiceImpl.java | 4 +-
...entmesh.openconnect.api.ConnectorCreateService} | 3 +-
.../{source-config.yml => sink-config.yml} | 7 +-
.../src/main/resources/source-config.yml | 1 -
.../apache/eventmesh/openconnect/Application.java | 27 +++-
...ateService.java => ConnectorCreateService.java} | 9 +-
.../apache/eventmesh/runtime/demo/CCSubClient.java | 1 -
19 files changed, 394 insertions(+), 69 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 9a1dc0686..41fe8dfa9 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
@@ -173,6 +173,10 @@ public class Constants {
public static final int SUCCESS_CODE = 200;
+ public static final String SINK = "Sink";
+
+ public static final String SOURCE = "Source";
+
// protocol desc
public static final String PROTOCOL_DESC_GRPC_CLOUD_EVENT =
"grpc-cloud-event";
diff --git
a/eventmesh-common/src/main/java/org/apache/eventmesh/common/utils/JsonUtils.java
b/eventmesh-common/src/main/java/org/apache/eventmesh/common/utils/JsonUtils.java
index 487ecb9b7..19614714f 100644
---
a/eventmesh-common/src/main/java/org/apache/eventmesh/common/utils/JsonUtils.java
+++
b/eventmesh-common/src/main/java/org/apache/eventmesh/common/utils/JsonUtils.java
@@ -24,6 +24,7 @@ import org.apache.eventmesh.common.exception.JsonException;
import org.apache.commons.lang3.StringUtils;
import java.io.IOException;
+import java.lang.reflect.Type;
import java.util.Objects;
import com.fasterxml.jackson.annotation.JsonInclude;
@@ -98,6 +99,24 @@ public class JsonUtils {
}
}
+ public static <T> T parseObject(String text, Type type) {
+ if (StringUtils.isEmpty(text)) {
+ return null;
+ }
+ try {
+ TypeReference<T> typeReference = new TypeReference<T>() {
+
+ @Override
+ public Type getType() {
+ return type;
+ }
+ };
+ return OBJECT_MAPPER.readValue(text, typeReference);
+ } catch (JsonProcessingException e) {
+ throw new JsonException("deserialize json string to object error",
e);
+ }
+ }
+
public static <T> T parseObject(byte[] bytes, Class<T> clazz) {
if (bytes == null || bytes.length == 0) {
return null;
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
index aa110b2aa..bb07baa94 100644
---
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
@@ -25,20 +25,21 @@ public class SpringApplicationContextHolder implements
ApplicationContextAware {
private static ApplicationContext applicationContext;
+ public static boolean isStarted() {
+ return applicationContext != null;
+ }
+
@Override
public void setApplicationContext(ApplicationContext applicationContext)
throws BeansException {
SpringApplicationContextHolder.applicationContext = applicationContext;
}
- public static Object getBean(String beanName) {
- return applicationContext.getBean(beanName);
+ public static <T> T getBean(Class<T> clazz) {
+ return applicationContext.getBean(clazz);
}
- public static Object getBean(Class<?> beanClass) {
- return applicationContext.getBean(beanClass);
+ public static Object getBean(String beanName) {
+ return applicationContext.getBean(beanName);
}
- 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
index 008efef8f..12f56a501 100644
---
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
@@ -19,6 +19,8 @@ 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.sink.EventMeshListenerBeanPostProcessor;
+import
org.apache.eventmesh.connector.spring.sink.connector.SpringSinkConnector;
import
org.apache.eventmesh.connector.spring.source.connector.SpringSourceConnector;
import
org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
@@ -32,8 +34,10 @@ import org.springframework.context.annotation.Configuration;
public class EventMeshAutoConfiguration {
public static final String SPRING_SOURCE_CONNECTOR_BEAN_NAME =
"springSourceConnector";
+ public static final String SPRING_SINK_CONNECTOR_BEAN_NAME =
"springSinkConnector";
public static final String SPRING_CONNECT_SERVER_BEAN_NAME =
"springConnectServer";
public static final String SPRING_APPLICATION_CONTEXT_HOLDER =
"springApplicationContextHolder";
+ public static final String EVENTMESH_LISTENER_BEAN_POST_PROCESSOR =
"eventMeshListenerBeanPostProcessor";
@Bean(name = SPRING_SOURCE_CONNECTOR_BEAN_NAME)
@ConditionalOnMissingBean(name = SPRING_SOURCE_CONNECTOR_BEAN_NAME)
@@ -41,6 +45,12 @@ public class EventMeshAutoConfiguration {
return new SpringSourceConnector();
}
+ @Bean(name = SPRING_SINK_CONNECTOR_BEAN_NAME)
+ @ConditionalOnMissingBean(name = SPRING_SINK_CONNECTOR_BEAN_NAME)
+ public SpringSinkConnector springSinkConnector() {
+ return new SpringSinkConnector();
+ }
+
@Bean(name = SPRING_CONNECT_SERVER_BEAN_NAME)
@ConditionalOnMissingBean(name = SPRING_CONNECT_SERVER_BEAN_NAME)
public SpringConnectServer springConnectServer() {
@@ -52,4 +62,11 @@ public class EventMeshAutoConfiguration {
public SpringApplicationContextHolder springApplicationContextHolder() {
return new SpringApplicationContextHolder();
}
+
+ @Bean(name = EVENTMESH_LISTENER_BEAN_POST_PROCESSOR)
+ @ConditionalOnMissingBean(name = EVENTMESH_LISTENER_BEAN_POST_PROCESSOR)
+ public EventMeshListenerBeanPostProcessor
eventMeshListenerBeanPostProcessor() {
+ return new EventMeshListenerBeanPostProcessor();
+ }
+
}
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
index 2c79b4b46..f176200ae 100644
---
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
@@ -23,17 +23,27 @@ import
org.apache.eventmesh.connector.spring.source.connector.SpringSourceConnec
import org.apache.eventmesh.openconnect.Application;
import org.apache.eventmesh.openconnect.util.ConfigUtil;
+import java.util.HashMap;
+import java.util.Map;
+
import org.springframework.boot.CommandLineRunner;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
public class SpringConnectServer implements CommandLineRunner {
+ private static final String SPRING_SOURCE = "springSource";
+
@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();
+ Map<String, String> extensions = new HashMap<>();
+ extensions.put(Application.CREATE_EXTENSION_KEY, SPRING_SOURCE);
+ Application application = new Application(extensions);
application.run(SpringSourceConnector.class);
}
}
diff --git
a/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/api/source/SourceCreateService.java
b/eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/sink/EventMeshConsumerMetadata.java
similarity index 70%
copy from
eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/api/source/SourceCreateService.java
copy to
eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/sink/EventMeshConsumerMetadata.java
index 8ffc0bb5d..1585480c8 100644
---
a/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/api/source/SourceCreateService.java
+++
b/eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/sink/EventMeshConsumerMetadata.java
@@ -15,16 +15,21 @@
* limitations under the License.
*/
-package org.apache.eventmesh.openconnect.api.source;
+package org.apache.eventmesh.connector.spring.sink;
-import org.apache.eventmesh.spi.EventMeshExtensionType;
-import org.apache.eventmesh.spi.EventMeshSPI;
+import java.lang.reflect.Method;
-/**
- * SPI interface for source connector creation.
- */
-@EventMeshSPI(eventMeshExtensionType = EventMeshExtensionType.CONNECTOR)
-public interface SourceCreateService {
+import lombok.AllArgsConstructor;
+import lombok.Data;
+
+@Data
+@AllArgsConstructor
+public class EventMeshConsumerMetadata {
+
+ private Object bean;
+
+ private Method method;
+
+ private EventMeshListener annotation;
- Source create();
}
diff --git
a/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/api/source/SourceCreateService.java
b/eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/sink/EventMeshListener.java
similarity index 60%
copy from
eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/api/source/SourceCreateService.java
copy to
eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/sink/EventMeshListener.java
index 8ffc0bb5d..e2a850a21 100644
---
a/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/api/source/SourceCreateService.java
+++
b/eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/sink/EventMeshListener.java
@@ -15,16 +15,24 @@
* limitations under the License.
*/
-package org.apache.eventmesh.openconnect.api.source;
+package org.apache.eventmesh.connector.spring.sink;
-import org.apache.eventmesh.spi.EventMeshExtensionType;
-import org.apache.eventmesh.spi.EventMeshSPI;
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
-/**
- * SPI interface for source connector creation.
- */
-@EventMeshSPI(eventMeshExtensionType = EventMeshExtensionType.CONNECTOR)
-public interface SourceCreateService {
+import org.springframework.stereotype.Component;
+
+@Target(ElementType.METHOD)
+@Retention(RetentionPolicy.RUNTIME)
+@Documented
+@Component
+public @interface EventMeshListener {
- Source create();
+ /**
+ * The requestTimeout of client,it is 5s by default.
+ */
+ int requestTimeout() default 5;
}
diff --git
a/eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/sink/EventMeshListenerBeanPostProcessor.java
b/eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/sink/EventMeshListenerBeanPostProcessor.java
new file mode 100644
index 000000000..b36e30841
--- /dev/null
+++
b/eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/sink/EventMeshListenerBeanPostProcessor.java
@@ -0,0 +1,143 @@
+/*
+ * 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.sink;
+
+import org.apache.eventmesh.common.Constants;
+import org.apache.eventmesh.common.ThreadPoolFactory;
+import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.connector.spring.config.SpringConnectServerConfig;
+import
org.apache.eventmesh.connector.spring.sink.connector.SpringSinkConnector;
+import org.apache.eventmesh.openconnect.Application;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.ConnectRecord;
+import org.apache.eventmesh.openconnect.util.ConfigUtil;
+
+import java.lang.reflect.Method;
+import java.lang.reflect.Type;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+import org.springframework.aop.support.AopUtils;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.config.BeanPostProcessor;
+import org.springframework.boot.CommandLineRunner;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationContextAware;
+import org.springframework.core.annotation.AnnotatedElementUtils;
+import org.springframework.util.ReflectionUtils;
+
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class EventMeshListenerBeanPostProcessor implements
ApplicationContextAware,
+ CommandLineRunner, BeanPostProcessor {
+
+ private static final String SPRING_SINK = "springSink";
+
+ private static final ThreadPoolExecutor executor =
ThreadPoolFactory.createThreadPoolExecutor(
+ Runtime.getRuntime().availableProcessors() * 2,
+ Runtime.getRuntime().availableProcessors() * 2,
+ "EventMesh-MessageListenerBeanPostProcessor-");
+
+ private ApplicationContext applicationContext;
+
+ private List<EventMeshConsumerMetadata> metadataList = new ArrayList<>();
+
+ @Override
+ public void setApplicationContext(ApplicationContext applicationContext)
throws BeansException {
+ this.applicationContext = applicationContext;
+ }
+
+ @Override
+ public Object postProcessAfterInitialization(Object bean, String beanName)
throws BeansException {
+ Class<?> targetClass = AopUtils.isAopProxy(bean) ?
AopUtils.getTargetClass(bean) : bean.getClass();
+ ReflectionUtils.doWithMethods(targetClass, new
ReflectionUtils.MethodCallback() {
+
+ @Override
+ public void doWith(final Method method) throws
IllegalArgumentException, IllegalAccessException {
+ EventMeshListener annotation =
AnnotatedElementUtils.findMergedAnnotation(method, EventMeshListener.class);
+ if (annotation == null || method.isBridge()) {
+ return;
+ }
+ metadataList.add(new EventMeshConsumerMetadata(bean, method,
annotation));
+ }
+ });
+ return bean;
+ }
+
+ @Override
+ public void run(String... args) throws Exception {
+ runSinkConnector();
+ metadataList.forEach(metadata -> {
+ Object bean = metadata.getBean();
+ Method method = metadata.getMethod();
+ EventMeshListener annotation = metadata.getAnnotation();
+ SpringSinkConnector sinkConnector =
applicationContext.getBean(SpringSinkConnector.class);
+ executor.execute(() -> {
+ ConnectRecord poll;
+ while (sinkConnector.isRunning()) {
+ try {
+ poll =
sinkConnector.getQueue().poll(annotation.requestTimeout(), TimeUnit.SECONDS);
+ if (null == poll || null == poll.getData()) {
+ continue;
+ }
+ String messageBody = new String((byte[])
poll.getData());
+ Type[] parameterizedTypes =
method.getGenericParameterTypes();
+ if (parameterizedTypes.length == 0) {
+ throw new IllegalStateException("There has not any
arguments for consumer method.");
+ }
+ if (parameterizedTypes.length > 1) {
+ throw new IllegalStateException("There has more
than one arguments for consumer method.");
+ }
+ Class<?> rawType;
+ if (parameterizedTypes[0] instanceof Class) {
+ rawType = (Class<?>) parameterizedTypes[0];
+ } else {
+ throw new IllegalStateException(
+ "The arguments type for consumer method can't
cast to be Class and ParameterizedTypeImpl");
+ }
+ if (rawType == String.class) {
+ metadata.getMethod().invoke(bean, messageBody);
+ } else {
+ metadata.getMethod().invoke(bean,
JsonUtils.parseObject(messageBody, parameterizedTypes[0]));
+ }
+ } catch (Exception e) {
+ log.warn("Consume snapshot event error", e);
+ }
+ }
+ });
+ });
+ }
+
+ @SneakyThrows
+ public void runSinkConnector() {
+ SpringConnectServerConfig springConnectServerConfig =
ConfigUtil.parse(SpringConnectServerConfig.class,
+ Constants.CONNECT_SERVER_CONFIG_FILE_NAME);
+
+ if (springConnectServerConfig.isSinkEnable()) {
+ Map<String, String> extensions = new HashMap<>();
+ extensions.put(Application.CREATE_EXTENSION_KEY, SPRING_SINK);
+ Application application = new Application(extensions);
+ application.run(SpringSinkConnector.class);
+ }
+ }
+}
diff --git
a/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/api/source/SourceCreateService.java
b/eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/sink/config/SpringSinkConfig.java
similarity index 66%
copy from
eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/api/source/SourceCreateService.java
copy to
eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/sink/config/SpringSinkConfig.java
index 8ffc0bb5d..aff0d8a0f 100644
---
a/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/api/source/SourceCreateService.java
+++
b/eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/sink/config/SpringSinkConfig.java
@@ -15,16 +15,17 @@
* limitations under the License.
*/
-package org.apache.eventmesh.openconnect.api.source;
+package org.apache.eventmesh.connector.spring.sink.config;
-import org.apache.eventmesh.spi.EventMeshExtensionType;
-import org.apache.eventmesh.spi.EventMeshSPI;
+import
org.apache.eventmesh.connector.spring.sink.connector.SinkConnectorConfig;
+import org.apache.eventmesh.openconnect.api.config.SinkConfig;
-/**
- * SPI interface for source connector creation.
- */
-@EventMeshSPI(eventMeshExtensionType = EventMeshExtensionType.CONNECTOR)
-public interface SourceCreateService {
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+@Data
+@EqualsAndHashCode(callSuper = true)
+public class SpringSinkConfig extends SinkConfig {
- Source create();
+ private SinkConnectorConfig sinkConnectorConfig;
}
diff --git
a/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/api/source/SourceCreateService.java
b/eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/sink/connector/SinkConnectorConfig.java
similarity index 70%
copy from
eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/api/source/SourceCreateService.java
copy to
eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/sink/connector/SinkConnectorConfig.java
index 8ffc0bb5d..a71c4ab66 100644
---
a/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/api/source/SourceCreateService.java
+++
b/eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/sink/connector/SinkConnectorConfig.java
@@ -15,16 +15,13 @@
* limitations under the License.
*/
-package org.apache.eventmesh.openconnect.api.source;
+package org.apache.eventmesh.connector.spring.sink.connector;
-import org.apache.eventmesh.spi.EventMeshExtensionType;
-import org.apache.eventmesh.spi.EventMeshSPI;
+import lombok.Data;
-/**
- * SPI interface for source connector creation.
- */
-@EventMeshSPI(eventMeshExtensionType = EventMeshExtensionType.CONNECTOR)
-public interface SourceCreateService {
+@Data
+public class SinkConnectorConfig {
+
+ private String connectorName;
- Source create();
}
diff --git
a/eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/sink/connector/SpringSinkConnector.java
b/eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/sink/connector/SpringSinkConnector.java
new file mode 100644
index 000000000..648e1f807
--- /dev/null
+++
b/eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/sink/connector/SpringSinkConnector.java
@@ -0,0 +1,106 @@
+/*
+ * 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.sink.connector;
+
+import org.apache.eventmesh.connector.spring.sink.config.SpringSinkConfig;
+import org.apache.eventmesh.openconnect.api.config.Config;
+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 java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * Spring sink connector
+ */
+@Slf4j
+public class SpringSinkConnector implements Sink {
+
+ private SpringSinkConfig sinkConfig;
+
+ private BlockingQueue<ConnectRecord> queue;
+
+ private volatile boolean isRunning = false;
+
+ @Override
+ public Class<? extends Config> configClass() {
+ return SpringSinkConfig.class;
+ }
+
+ @Override
+ public void init(Config config) throws Exception {
+ // init config for spring sink connector
+ this.sinkConfig = (SpringSinkConfig) config;
+ this.queue = new LinkedBlockingQueue<>(1000);
+ }
+
+ @Override
+ public void init(ConnectorContext connectorContext) throws Exception {
+ // init config for openfunction source connector
+ SinkConnectorContext sinkConnectorContext = (SinkConnectorContext)
connectorContext;
+ this.sinkConfig = (SpringSinkConfig)
sinkConnectorContext.getSinkConfig();
+ this.queue = new LinkedBlockingQueue<>(1000);
+ }
+
+ @Override
+ public void start() throws Exception {
+ isRunning = true;
+ }
+
+ @Override
+ public void commit(ConnectRecord record) {
+
+ }
+
+ @Override
+ public String name() {
+ return this.sinkConfig.getSinkConnectorConfig().getConnectorName();
+ }
+
+ @Override
+ public void stop() throws Exception {
+
+ }
+
+ public boolean isRunning() {
+ return isRunning;
+ }
+
+ public BlockingQueue<ConnectRecord> getQueue() {
+ return queue;
+ }
+
+ @Override
+ public void put(List<ConnectRecord> sinkRecords) {
+ for (ConnectRecord connectRecord : sinkRecords) {
+ try {
+ queue.put(connectRecord);
+ } catch (InterruptedException e) {
+ Thread currentThread = Thread.currentThread();
+ log.warn("[SpringSinkConnector] Interrupting thread {} due to
exception {}",
+ currentThread.getName(), e.getMessage());
+ currentThread.interrupt();
+ }
+ }
+ }
+}
diff --git
a/eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/source/connector/SpringSourceConnectorCreateServiceImpl.java
b/eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/sink/connector/SpringSinkConnectorCreateServiceImpl.java
similarity index 71%
copy from
eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/source/connector/SpringSourceConnectorCreateServiceImpl.java
copy to
eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/sink/connector/SpringSinkConnectorCreateServiceImpl.java
index 0f61caebc..e7e7b2554 100644
---
a/eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/source/connector/SpringSourceConnectorCreateServiceImpl.java
+++
b/eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/sink/connector/SpringSinkConnectorCreateServiceImpl.java
@@ -15,19 +15,19 @@
* limitations under the License.
*/
-package org.apache.eventmesh.connector.spring.source.connector;
+package org.apache.eventmesh.connector.spring.sink.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;
+import org.apache.eventmesh.openconnect.api.ConnectorCreateService;
+import org.apache.eventmesh.openconnect.api.sink.Sink;
-public class SpringSourceConnectorCreateServiceImpl implements
SourceCreateService {
+public class SpringSinkConnectorCreateServiceImpl implements
ConnectorCreateService<Sink> {
@Override
- public Source create() {
+ public Sink create() {
if (SpringApplicationContextHolder.isStarted()) {
- return (Source)
SpringApplicationContextHolder.getBean(EventMeshAutoConfiguration.SPRING_SOURCE_CONNECTOR_BEAN_NAME);
+ return (Sink)
SpringApplicationContextHolder.getBean(EventMeshAutoConfiguration.SPRING_SINK_CONNECTOR_BEAN_NAME);
}
return null;
}
diff --git
a/eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/source/connector/SpringSourceConnectorCreateServiceImpl.java
b/eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/source/connector/SpringSourceConnectorCreateServiceImpl.java
index 0f61caebc..573ff9ed7 100644
---
a/eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/source/connector/SpringSourceConnectorCreateServiceImpl.java
+++
b/eventmesh-connectors/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/source/connector/SpringSourceConnectorCreateServiceImpl.java
@@ -19,10 +19,10 @@ 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.ConnectorCreateService;
import org.apache.eventmesh.openconnect.api.source.Source;
-import org.apache.eventmesh.openconnect.api.source.SourceCreateService;
-public class SpringSourceConnectorCreateServiceImpl implements
SourceCreateService {
+public class SpringSourceConnectorCreateServiceImpl implements
ConnectorCreateService<Source> {
@Override
public Source create() {
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.ConnectorCreateService
similarity index 79%
rename from
eventmesh-connectors/eventmesh-connector-spring/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.openconnect.api.source.SourceCreateService
rename to
eventmesh-connectors/eventmesh-connector-spring/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.openconnect.api.ConnectorCreateService
index b79216363..1daf7796f 100644
---
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.ConnectorCreateService
@@ -15,4 +15,5 @@
# limitations under the License.
#
-spring=org.apache.eventmesh.connector.spring.source.connector.SpringSourceConnectorCreateServiceImpl
\ No newline at end of file
+springSink=org.apache.eventmesh.connector.spring.sink.connector.SpringSinkConnectorCreateServiceImpl
+springSource=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/source-config.yml
b/eventmesh-connectors/eventmesh-connector-spring/src/main/resources/sink-config.yml
similarity index 89%
copy from
eventmesh-connectors/eventmesh-connector-spring/src/main/resources/source-config.yml
copy to
eventmesh-connectors/eventmesh-connector-spring/src/main/resources/sink-config.yml
index a0de4f218..dd5c081af 100644
---
a/eventmesh-connectors/eventmesh-connector-spring/src/main/resources/source-config.yml
+++
b/eventmesh-connectors/eventmesh-connector-spring/src/main/resources/sink-config.yml
@@ -20,10 +20,9 @@ pubSubConfig:
subject: TEST-TOPIC-SPRING
idc: FT
env: PRD
- group: springSource
+ group: springSink
appId: 5033
- userName: springSourceUser
+ userName: springSinkUser
passWord: springPassWord
connectorConfig:
- connectorName: springSource
- topic: SourceTopic
+ connectorName: springSink
\ No newline at end of file
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
index a0de4f218..4559cd670 100644
---
a/eventmesh-connectors/eventmesh-connector-spring/src/main/resources/source-config.yml
+++
b/eventmesh-connectors/eventmesh-connector-spring/src/main/resources/source-config.yml
@@ -26,4 +26,3 @@ pubSubConfig:
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 740bd1226..3252a52e3 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
@@ -17,16 +17,18 @@
package org.apache.eventmesh.openconnect;
+import org.apache.eventmesh.openconnect.api.ConnectorCreateService;
import org.apache.eventmesh.openconnect.api.config.Config;
import org.apache.eventmesh.openconnect.api.config.SinkConfig;
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 org.apache.commons.collections4.MapUtils;
+
import java.util.HashMap;
import java.util.Map;
@@ -37,16 +39,29 @@ public class Application {
public static final Map<String, Connector> CONNECTOR_MAP = new HashMap<>();
- private static final String SPRING = "spring";
+ public static final String CREATE_EXTENSION_KEY = "createExtension";
+
+ private Map<String, String> extensions;
+
+ public Application() {
+
+ }
+
+ public Application(Map<String, String> extensions) {
+ this.extensions = extensions;
+ }
public void run(Class<? extends Connector> clazz) throws Exception {
Connector connector = null;
try {
- SourceCreateService createService =
-
EventMeshExtensionFactory.getExtension(SourceCreateService.class, SPRING);
- if (createService != null) {
- connector = createService.create();
+ if (MapUtils.isNotEmpty(extensions) &&
extensions.containsKey(CREATE_EXTENSION_KEY)) {
+ String spiKey = extensions.get(CREATE_EXTENSION_KEY);
+ ConnectorCreateService<?> createService =
+
EventMeshExtensionFactory.getExtension(ConnectorCreateService.class, spiKey);
+ if (createService != null) {
+ connector = createService.create();
+ }
}
if (connector == null) {
connector = clazz.getDeclaredConstructor().newInstance();
diff --git
a/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/api/source/SourceCreateService.java
b/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/api/ConnectorCreateService.java
similarity index 81%
rename from
eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/api/source/SourceCreateService.java
rename to
eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/api/ConnectorCreateService.java
index 8ffc0bb5d..15e645474 100644
---
a/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/api/source/SourceCreateService.java
+++
b/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/api/ConnectorCreateService.java
@@ -15,16 +15,17 @@
* limitations under the License.
*/
-package org.apache.eventmesh.openconnect.api.source;
+package org.apache.eventmesh.openconnect.api;
+import org.apache.eventmesh.openconnect.api.connector.Connector;
import org.apache.eventmesh.spi.EventMeshExtensionType;
import org.apache.eventmesh.spi.EventMeshSPI;
/**
- * SPI interface for source connector creation.
+ * SPI interface for connector creation.
*/
@EventMeshSPI(eventMeshExtensionType = EventMeshExtensionType.CONNECTOR)
-public interface SourceCreateService {
+public interface ConnectorCreateService<T extends Connector> {
- Source create();
+ T create();
}
diff --git
a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/demo/CCSubClient.java
b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/demo/CCSubClient.java
index e16108a7b..e7cba1d22 100644
---
a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/demo/CCSubClient.java
+++
b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/demo/CCSubClient.java
@@ -25,7 +25,6 @@ import
org.apache.eventmesh.runtime.client.common.MessageUtils;
import org.apache.eventmesh.runtime.client.common.UserAgentUtils;
import org.apache.eventmesh.runtime.client.impl.SubClientImpl;
-
import lombok.extern.slf4j.Slf4j;
@Slf4j
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
