This is an automated email from the ASF dual-hosted git repository.
jonyang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-eventmesh.git
The following commit(s) were added to refs/heads/master by this push:
new 4c45d6967 [ISSUE #3184]Refactor ConnectorResource (#3190)
4c45d6967 is described below
commit 4c45d69676bc935da3fd0ab05ebaf83f25ada764
Author: mxsm <[email protected]>
AuthorDate: Sat Feb 18 07:43:31 2023 +0800
[ISSUE #3184]Refactor ConnectorResource (#3190)
---
.../api/connector/ConnectorResourceService.java | 4 +-
.../eventmesh/runtime/boot/EventMeshServer.java | 4 +-
.../runtime/connector/ConnectorResource.java | 46 +++++++++++++++++++---
3 files changed, 44 insertions(+), 10 deletions(-)
diff --git
a/eventmesh-connector-plugin/eventmesh-connector-api/src/main/java/org/apache/eventmesh/api/connector/ConnectorResourceService.java
b/eventmesh-connector-plugin/eventmesh-connector-api/src/main/java/org/apache/eventmesh/api/connector/ConnectorResourceService.java
index 2657fe5ab..edb61653f 100644
---
a/eventmesh-connector-plugin/eventmesh-connector-api/src/main/java/org/apache/eventmesh/api/connector/ConnectorResourceService.java
+++
b/eventmesh-connector-plugin/eventmesh-connector-api/src/main/java/org/apache/eventmesh/api/connector/ConnectorResourceService.java
@@ -27,14 +27,14 @@ import org.apache.eventmesh.spi.EventMeshSPI;
public interface ConnectorResourceService {
/**
- * Resource initialization in connector,such as,some public threadpool if
exist
+ * Resource initialization in connector,such as,some public thread pool if
exist
*
* @throws Exception
*/
void init() throws Exception;
/**
- * Resource release in connector,such as,some public threadpool if exist
+ * Resource release in connector,such as,some public thread pool if exist
*
* @throws Exception
*/
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshServer.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshServer.java
index 6c4900c3d..34e8785c8 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshServer.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshServer.java
@@ -65,7 +65,7 @@ public class EventMeshServer {
this.registry = new Registry();
trace = new Trace(this.configuration.isEventMeshServerTraceEnable());
- this.connectorResource = new ConnectorResource();
+ this.connectorResource =
ConnectorResource.getInstance(this.configuration.getEventMeshConnectorPluginType());
final List<String> provideServerProtocols =
configuration.getEventMeshProvideServerProtocols();
for (final String provideServerProtocol : provideServerProtocols) {
@@ -84,7 +84,7 @@ public class EventMeshServer {
public void init() throws Exception {
if (Objects.nonNull(configuration)) {
-
connectorResource.init(configuration.getEventMeshConnectorPluginType());
+ connectorResource.init();
if (configuration.isEventMeshServerSecurityEnable()) {
acl.init();
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/connector/ConnectorResource.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/connector/ConnectorResource.java
index b53969c7a..71bef8eab 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/connector/ConnectorResource.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/connector/ConnectorResource.java
@@ -21,23 +21,57 @@ import
org.apache.eventmesh.api.connector.ConnectorResourceService;
import org.apache.eventmesh.spi.EventMeshExtensionFactory;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+
import lombok.extern.slf4j.Slf4j;
@Slf4j
public class ConnectorResource {
- private static ConnectorResourceService connectorResourceService;
+ private static final Map<String, ConnectorResource>
CONNECTOR_RESOURCE_CACHE = new HashMap<>(16);
+
+ private ConnectorResourceService connectorResourceService;
+
+ private final AtomicBoolean inited = new AtomicBoolean(false);
+
+ private final AtomicBoolean released = new AtomicBoolean(false);
+
+ private ConnectorResource() {
+
+ }
+
+ public static ConnectorResource getInstance(String
connectorResourcePluginType) {
+ return
CONNECTOR_RESOURCE_CACHE.computeIfAbsent(connectorResourcePluginType, key ->
connectorResourceBuilder(key));
+ }
- public void init(String connectorResourcePluginType) throws Exception {
- connectorResourceService =
EventMeshExtensionFactory.getExtension(ConnectorResourceService.class,
connectorResourcePluginType);
- if (connectorResourceService == null) {
- log.error("can't load the connectorResourceService plugin, please
check.");
- throw new RuntimeException("doesn't load the
connectorResourceService plugin, please check.");
+ private static ConnectorResource connectorResourceBuilder(String
connectorResourcePluginType) {
+ ConnectorResourceService connectorResourceServiceExt =
EventMeshExtensionFactory.getExtension(ConnectorResourceService.class,
+ connectorResourcePluginType);
+ if (connectorResourceServiceExt == null) {
+ String errorMsg = "can't load the connectorResourceService plugin,
please check.";
+ log.error(errorMsg);
+ throw new RuntimeException(errorMsg);
+ }
+ ConnectorResource connectorResource = new ConnectorResource();
+ connectorResource.connectorResourceService =
connectorResourceServiceExt;
+ return connectorResource;
+ }
+
+ public void init() throws Exception {
+ if (!inited.compareAndSet(false, true)) {
+ return;
}
connectorResourceService.init();
}
public void release() throws Exception {
+ if (!released.compareAndSet(false, true)) {
+ return;
+ }
+ inited.compareAndSet(true, false);
connectorResourceService.release();
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
