This is an automated email from the ASF dual-hosted git repository.
xiatian pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/eventmesh-dashboard.git
The following commit(s) were added to refs/heads/main by this push:
new 13d3728 [ISSUE #97] Runtime tcp http grpc client impl (#96)
13d3728 is described below
commit 13d372821644467efb3a2214d0016e88f5b8e477
Author: Alonexc <[email protected]>
AuthorDate: Mon Apr 15 17:45:02 2024 +0800
[ISSUE #97] Runtime tcp http grpc client impl (#96)
* runtime tcp http grpc client impl
* check
* check
* check
* check
* check
* check
* check
* check
* Upgrade the grpc version.
* check
* check
* ci try-catch
* ci try-catch
* redis client update
* redis client update
* redis client test
* ci
* ci
* etcd add connect time
* update
* optimisation
---
eventmesh-dashboard-core/pom.xml | 30 +++++-
.../dashboard/core/function/SDK/SDKManager.java | 35 ++++++-
.../dashboard/core/function/SDK/SDKTypeEnum.java | 12 +++
.../core/function/SDK/config/CreateEtcdConfig.java | 9 ++
.../function/SDK/config/CreateRedisConfig.java | 11 +++
.../function/SDK/config/CreateRuntimeConfig.java | 70 ++++++++++++++
.../function/SDK/operation/EtcdSDKOperation.java | 4 +-
.../function/SDK/operation/NacosSDKOperation.java | 5 +-
.../function/SDK/operation/RedisSDKOperation.java | 15 ++-
...n.java => RuntimeGrpcConsumerSDKOperation.java} | 36 +++----
...n.java => RuntimeGrpcProducerSDKOperation.java} | 35 +++----
...n.java => RuntimeHttpConsumerSDKOperation.java} | 35 +++----
...n.java => RuntimeHttpProducerSDKOperation.java} | 35 +++----
.../SDK/operation/RuntimeSDKOperation.java | 100 ++++++++++++++++++-
.../RuntimeTcpCloudEventSDKOperation.java | 62 ++++++++++++
.../operation/RuntimeTcpEventMeshSDKOperation.java | 62 ++++++++++++
.../RuntimeTcpOpenMessageSDKOperation.java | 57 +++++++++++
.../SDK/util/RuntimeSDKOperationUtils.java | 107 +++++++++++++++++++++
.../core/function/SDK/wrapper/NacosSDKWrapper.java | 2 +
.../function/SDK/wrapper/RuntimeSDKWrapper.java | 71 ++++++++++++++
.../core/function/SDK/SDKManagerTest.java | 8 +-
.../SDK/operation/EtcdSDKCreateOperationTest.java | 34 +++----
.../SDK/operation/RedisSDKCreateOperationTest.java | 19 ++--
...va => RuntimeGrpcConsumerSDKOperationTest.java} | 37 ++++---
...va => RuntimeGrpcProducerSDKOperationTest.java} | 37 ++++---
.../RuntimeHttpConsumerSDKOperationTest.java | 62 ++++++++++++
.../RuntimeHttpProducerSDKOperationTest.java | 62 ++++++++++++
.../SDK/operation/RuntimeSDKOperationTest.java | 73 ++++++++++++++
.../RuntimeTcpCloudEventSDKOperationTest.java | 71 ++++++++++++++
.../RuntimeTcpEventMeshSDKOperationTest.java | 71 ++++++++++++++
.../RuntimeTcpOpenMessageSDKOperationTest.java} | 17 ++--
31 files changed, 1120 insertions(+), 164 deletions(-)
diff --git a/eventmesh-dashboard-core/pom.xml b/eventmesh-dashboard-core/pom.xml
index ca13340..9f3a19c 100644
--- a/eventmesh-dashboard-core/pom.xml
+++ b/eventmesh-dashboard-core/pom.xml
@@ -40,6 +40,19 @@
</properties>
<dependencies>
+ <!-- grpc -->
+ <dependency>
+ <groupId>io.grpc</groupId>
+ <artifactId>grpc-all</artifactId>
+ <version>1.51.0</version>
+ </dependency>
+ <!-- guava -->
+ <dependency>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava</artifactId>
+ <version>32.1.3-jre</version>
+ </dependency>
+
<!-- EventMesh Dashboard modules -->
<dependency>
<groupId>org.apache.eventmesh.dashboard.common</groupId>
@@ -52,6 +65,19 @@
<version>0.0.1-SNAPSHOT</version>
</dependency>
+ <!-- eventmesh -->
+ <dependency>
+ <groupId>org.apache.eventmesh</groupId>
+ <artifactId>eventmesh-sdk-java</artifactId>
+ <version>1.10.0-release</version>
+ <exclusions>
+ <exclusion>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-slf4j-impl</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+
<!-- Meta - nacos client -->
<dependency>
<groupId>com.alibaba.nacos</groupId>
@@ -61,7 +87,7 @@
<dependency>
<groupId>io.etcd</groupId>
<artifactId>jetcd-core</artifactId>
- <version>0.3.0</version>
+ <version>0.7.5</version>
</dependency>
<!-- health check client -->
@@ -121,4 +147,4 @@
<scope>test</scope>
</dependency>
</dependencies>
-</project>
\ No newline at end of file
+</project>
diff --git
a/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/SDKManager.java
b/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/SDKManager.java
index a203ff3..e8e722e 100644
---
a/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/SDKManager.java
+++
b/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/SDKManager.java
@@ -26,6 +26,14 @@ import
org.apache.eventmesh.dashboard.core.function.SDK.operation.RedisSDKOperat
import
org.apache.eventmesh.dashboard.core.function.SDK.operation.RocketMQProduceSDKOperation;
import
org.apache.eventmesh.dashboard.core.function.SDK.operation.RocketMQPushConsumerSDKOperation;
import
org.apache.eventmesh.dashboard.core.function.SDK.operation.RocketMQRemotingSDKOperation;
+import
org.apache.eventmesh.dashboard.core.function.SDK.operation.RuntimeGrpcConsumerSDKOperation;
+import
org.apache.eventmesh.dashboard.core.function.SDK.operation.RuntimeGrpcProducerSDKOperation;
+import
org.apache.eventmesh.dashboard.core.function.SDK.operation.RuntimeHttpConsumerSDKOperation;
+import
org.apache.eventmesh.dashboard.core.function.SDK.operation.RuntimeHttpProducerSDKOperation;
+import
org.apache.eventmesh.dashboard.core.function.SDK.operation.RuntimeSDKOperation;
+import
org.apache.eventmesh.dashboard.core.function.SDK.operation.RuntimeTcpCloudEventSDKOperation;
+import
org.apache.eventmesh.dashboard.core.function.SDK.operation.RuntimeTcpEventMeshSDKOperation;
+import
org.apache.eventmesh.dashboard.core.function.SDK.operation.RuntimeTcpOpenMessageSDKOperation;
import java.util.AbstractMap.SimpleEntry;
import java.util.Map;
@@ -38,10 +46,17 @@ import java.util.concurrent.ConcurrentHashMap;
*/
public class SDKManager {
- private static final SDKManager INSTANCE = new SDKManager();
+ private static volatile SDKManager INSTANCE = null;
- public static SDKManager getInstance() {
+ public static synchronized SDKManager getInstance() {
+ if (INSTANCE == null) {
+ synchronized (SDKManager.class) {
+ if (INSTANCE == null) {
+ INSTANCE = new SDKManager();
+ }
+ }
+ }
return INSTANCE;
}
@@ -73,6 +88,17 @@ public class SDKManager {
clientCreateOperationMap.put(SDKTypeEnum.META_ETCD, new
EtcdSDKOperation());
+ clientCreateOperationMap.put(SDKTypeEnum.RUNTIME_EVENTMESH_CLIENT, new
RuntimeSDKOperation());
+
+
clientCreateOperationMap.put(SDKTypeEnum.RUNTIME_TCP_CLOUDEVENT_CLIENT, new
RuntimeTcpCloudEventSDKOperation());
+ clientCreateOperationMap.put(SDKTypeEnum.RUNTIME_TCP_EVENTMESH_CLIENT,
new RuntimeTcpEventMeshSDKOperation());
+
clientCreateOperationMap.put(SDKTypeEnum.RUNTIME_TCP_OPENMESSAGE_CLIENT, new
RuntimeTcpOpenMessageSDKOperation());
+
+ clientCreateOperationMap.put(SDKTypeEnum.RUNTIME_HTTP_PRODUCER, new
RuntimeHttpProducerSDKOperation());
+ clientCreateOperationMap.put(SDKTypeEnum.RUNTIME_HTTP_CONSUMER, new
RuntimeHttpConsumerSDKOperation());
+
+ clientCreateOperationMap.put(SDKTypeEnum.RUNTIME_GRPC_PRODUCER, new
RuntimeGrpcProducerSDKOperation());
+ clientCreateOperationMap.put(SDKTypeEnum.RUNTIME_GRPC_CONSUMER, new
RuntimeGrpcConsumerSDKOperation());
}
private SDKManager() {
@@ -114,4 +140,9 @@ public class SDKManager {
public Object getClient(SDKTypeEnum clientTypeEnum, String uniqueKey) {
return this.clientMap.get(clientTypeEnum).get(uniqueKey);
}
+
+ // get all client
+ public Map<String, Object> getClients(SDKTypeEnum clientTypeEnum) {
+ return this.clientMap.get(clientTypeEnum);
+ }
}
diff --git
a/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/SDKTypeEnum.java
b/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/SDKTypeEnum.java
index c6524c8..b10cde9 100644
---
a/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/SDKTypeEnum.java
+++
b/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/SDKTypeEnum.java
@@ -35,4 +35,16 @@ public enum SDKTypeEnum {
META_NACOS_NAMING,
META_ETCD,
+
+ RUNTIME_EVENTMESH_CLIENT,
+
+ RUNTIME_TCP_CLOUDEVENT_CLIENT,
+ RUNTIME_TCP_EVENTMESH_CLIENT,
+ RUNTIME_TCP_OPENMESSAGE_CLIENT,
+
+ RUNTIME_HTTP_PRODUCER,
+ RUNTIME_HTTP_CONSUMER,
+
+ RUNTIME_GRPC_PRODUCER,
+ RUNTIME_GRPC_CONSUMER,
}
diff --git
a/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/config/CreateEtcdConfig.java
b/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/config/CreateEtcdConfig.java
index 1168a65..2f116e0 100644
---
a/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/config/CreateEtcdConfig.java
+++
b/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/config/CreateEtcdConfig.java
@@ -17,13 +17,22 @@
package org.apache.eventmesh.dashboard.core.function.SDK.config;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
import lombok.Data;
+import lombok.NoArgsConstructor;
@Data
+@Builder
+@AllArgsConstructor
+@NoArgsConstructor
public class CreateEtcdConfig implements CreateSDKConfig {
private String etcdServerAddress;
+ @Builder.Default()
+ private int connectTime = 10;
+
@Override
public String getUniqueKey() {
return etcdServerAddress;
diff --git
a/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/config/CreateRedisConfig.java
b/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/config/CreateRedisConfig.java
index a457749..fb42f88 100644
---
a/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/config/CreateRedisConfig.java
+++
b/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/config/CreateRedisConfig.java
@@ -17,13 +17,24 @@
package org.apache.eventmesh.dashboard.core.function.SDK.config;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
import lombok.Data;
+import lombok.NoArgsConstructor;
@Data
+@Builder
+@AllArgsConstructor
+@NoArgsConstructor
public class CreateRedisConfig implements CreateSDKConfig {
private String redisUrl;
+ private String password;
+
+ @Builder.Default
+ private int timeOut = 10;
+
@Override
public String getUniqueKey() {
return redisUrl;
diff --git
a/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/config/CreateRuntimeConfig.java
b/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/config/CreateRuntimeConfig.java
new file mode 100644
index 0000000..7ddfc80
--- /dev/null
+++
b/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/config/CreateRuntimeConfig.java
@@ -0,0 +1,70 @@
+/*
+ * 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.dashboard.core.function.SDK.config;
+
+import org.apache.eventmesh.common.protocol.tcp.UserAgent;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@Builder
+@AllArgsConstructor
+@NoArgsConstructor
+public class CreateRuntimeConfig implements CreateSDKConfig {
+
+ // 127.0.0.1:10105;127.0.0.2:10105
+ private String runtimeServerAddress;
+
+ // protocol example:HTTP,TCP,GRPC
+ private String protocol;
+ // for tcp:cloudevents,eventmeshmessage,openmessage
+ private String protocolName;
+
+ // producer or consumer
+ private String clientType;
+
+ // topic
+ private String topic;
+
+ // config
+ private String env;
+ private String idc;
+ private String ip;
+ private String sys;
+ private String pid;
+ private String username;
+ private String password;
+
+ // producer
+ private String producerGroup;
+
+ // consumer
+ private String consumerGroup;
+ private String subUrl;
+
+ // Agent
+ private UserAgent userAgent;
+
+ @Override
+ public String getUniqueKey() {
+ return runtimeServerAddress;
+ }
+}
\ No newline at end of file
diff --git
a/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/EtcdSDKOperation.java
b/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/EtcdSDKOperation.java
index b630a38..98f88bd 100644
---
a/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/EtcdSDKOperation.java
+++
b/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/EtcdSDKOperation.java
@@ -21,6 +21,7 @@ import
org.apache.eventmesh.dashboard.core.function.SDK.AbstractSDKOperation;
import
org.apache.eventmesh.dashboard.core.function.SDK.config.CreateEtcdConfig;
import org.apache.eventmesh.dashboard.core.function.SDK.config.CreateSDKConfig;
+import java.time.Duration;
import java.util.AbstractMap.SimpleEntry;
import io.etcd.jetcd.Client;
@@ -39,6 +40,7 @@ public class EtcdSDKOperation extends
AbstractSDKOperation<KV> {
try {
final Client client = Client.builder()
.endpoints(getSplitEndpoints(etcdConfig))
+
.connectTimeout(Duration.ofSeconds(etcdConfig.getConnectTime()))
.build();
kvClient = client.getKVClient();
} catch (EtcdException e) {
@@ -48,7 +50,7 @@ public class EtcdSDKOperation extends
AbstractSDKOperation<KV> {
}
private static String[] getSplitEndpoints(CreateEtcdConfig etcdConfig) {
- return etcdConfig.getEtcdServerAddress().split(",");
+ return etcdConfig.getEtcdServerAddress().split(";");
}
@Override
diff --git
a/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/NacosSDKOperation.java
b/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/NacosSDKOperation.java
index c3651f6..ba78cbb 100644
---
a/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/NacosSDKOperation.java
+++
b/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/NacosSDKOperation.java
@@ -22,6 +22,7 @@ import
org.apache.eventmesh.dashboard.core.function.SDK.config.CreateSDKConfig;
import
org.apache.eventmesh.dashboard.core.function.SDK.wrapper.NacosSDKWrapper;
import java.util.AbstractMap.SimpleEntry;
+import java.util.Objects;
import com.alibaba.nacos.api.config.ConfigService;
import com.alibaba.nacos.api.naming.NamingService;
@@ -34,8 +35,8 @@ public class NacosSDKOperation extends
AbstractSDKOperation<NacosSDKWrapper> {
@Override
public SimpleEntry<String, NacosSDKWrapper> createClient(CreateSDKConfig
createClientConfig) {
SimpleEntry<String, ConfigService> configSimpleEntry =
nacosConfigClientCreateOperation.createClient(createClientConfig);
- SimpleEntry namingSimpleEntry =
nacosNamingClientCreateOperation.createClient(createClientConfig);
- if (configSimpleEntry.getKey() != namingSimpleEntry.getKey()) {
+ SimpleEntry<String, NamingService> namingSimpleEntry =
nacosNamingClientCreateOperation.createClient(createClientConfig);
+ if (!Objects.equals(configSimpleEntry.getKey(),
namingSimpleEntry.getKey())) {
throw new RuntimeException("Nacos config and naming server address
not match");
}
NacosSDKWrapper nacosClient = new NacosSDKWrapper(
diff --git
a/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RedisSDKOperation.java
b/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RedisSDKOperation.java
index 5fd3d4f..1717f54 100644
---
a/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RedisSDKOperation.java
+++
b/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RedisSDKOperation.java
@@ -21,17 +21,28 @@ import
org.apache.eventmesh.dashboard.core.function.SDK.AbstractSDKOperation;
import
org.apache.eventmesh.dashboard.core.function.SDK.config.CreateRedisConfig;
import org.apache.eventmesh.dashboard.core.function.SDK.config.CreateSDKConfig;
+import java.time.Duration;
import java.util.AbstractMap.SimpleEntry;
import io.lettuce.core.RedisClient;
+import io.lettuce.core.RedisURI;
import io.lettuce.core.api.StatefulRedisConnection;
public class RedisSDKOperation extends
AbstractSDKOperation<StatefulRedisConnection<String, String>> {
@Override
public SimpleEntry<String, StatefulRedisConnection<String, String>>
createClient(CreateSDKConfig clientConfig) {
- String redisUrl = ((CreateRedisConfig) clientConfig).getRedisUrl();
- RedisClient redisClient = RedisClient.create(redisUrl);
+ CreateRedisConfig redisConfig = (CreateRedisConfig) clientConfig;
+ String redisUrl = redisConfig.getRedisUrl();
+ String clientHost = redisUrl.split(":")[0];
+ int clientPort = Integer.parseInt(redisUrl.split(":")[1]);
+ RedisURI redisURI = RedisURI.builder()
+ .withHost(clientHost)
+ .withPort(clientPort)
+ .withPassword(redisConfig.getPassword())
+ .withTimeout(Duration.ofSeconds(redisConfig.getTimeOut()))
+ .build();
+ RedisClient redisClient = RedisClient.create(redisURI);
return new SimpleEntry<>(clientConfig.getUniqueKey(),
redisClient.connect());
}
diff --git
a/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/EtcdSDKOperation.java
b/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeGrpcConsumerSDKOperation.java
similarity index 54%
copy from
eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/EtcdSDKOperation.java
copy to
eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeGrpcConsumerSDKOperation.java
index b630a38..63e26a3 100644
---
a/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/EtcdSDKOperation.java
+++
b/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeGrpcConsumerSDKOperation.java
@@ -17,38 +17,34 @@
package org.apache.eventmesh.dashboard.core.function.SDK.operation;
+import static
org.apache.eventmesh.dashboard.core.function.SDK.util.RuntimeSDKOperationUtils.buildEventMeshGrpcConsumerConfig;
+
+import org.apache.eventmesh.client.grpc.config.EventMeshGrpcClientConfig;
+import org.apache.eventmesh.client.grpc.consumer.EventMeshGrpcConsumer;
+import org.apache.eventmesh.common.exception.EventMeshException;
import org.apache.eventmesh.dashboard.core.function.SDK.AbstractSDKOperation;
-import
org.apache.eventmesh.dashboard.core.function.SDK.config.CreateEtcdConfig;
+import
org.apache.eventmesh.dashboard.core.function.SDK.config.CreateRuntimeConfig;
import org.apache.eventmesh.dashboard.core.function.SDK.config.CreateSDKConfig;
import java.util.AbstractMap.SimpleEntry;
-import io.etcd.jetcd.Client;
-import io.etcd.jetcd.KV;
-import io.etcd.jetcd.common.exception.EtcdException;
-
import lombok.extern.slf4j.Slf4j;
@Slf4j
-public class EtcdSDKOperation extends AbstractSDKOperation<KV> {
+public class RuntimeGrpcConsumerSDKOperation extends
AbstractSDKOperation<EventMeshGrpcConsumer> {
@Override
- public SimpleEntry<String, KV> createClient(CreateSDKConfig clientConfig) {
- final CreateEtcdConfig etcdConfig = (CreateEtcdConfig) clientConfig;
- KV kvClient = null;
+ public SimpleEntry<String, EventMeshGrpcConsumer>
createClient(CreateSDKConfig clientConfig) {
+ final CreateRuntimeConfig runtimeConfig = (CreateRuntimeConfig)
clientConfig;
+ final EventMeshGrpcClientConfig grpcClientConfig =
buildEventMeshGrpcConsumerConfig(runtimeConfig);
+ EventMeshGrpcConsumer grpcConsumer = null;
try {
- final Client client = Client.builder()
- .endpoints(getSplitEndpoints(etcdConfig))
- .build();
- kvClient = client.getKVClient();
- } catch (EtcdException e) {
- log.error("create etcd client failed", e);
+ grpcConsumer = new EventMeshGrpcConsumer(grpcClientConfig);
+ grpcConsumer.init();
+ } catch (EventMeshException e) {
+ log.error("create runtime grpc Consumer client failed", e);
}
- return new SimpleEntry<>(clientConfig.getUniqueKey(), kvClient);
- }
-
- private static String[] getSplitEndpoints(CreateEtcdConfig etcdConfig) {
- return etcdConfig.getEtcdServerAddress().split(",");
+ return new SimpleEntry<>(clientConfig.getUniqueKey(), grpcConsumer);
}
@Override
diff --git
a/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/EtcdSDKOperation.java
b/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeGrpcProducerSDKOperation.java
similarity index 55%
copy from
eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/EtcdSDKOperation.java
copy to
eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeGrpcProducerSDKOperation.java
index b630a38..2c46a6b 100644
---
a/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/EtcdSDKOperation.java
+++
b/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeGrpcProducerSDKOperation.java
@@ -17,38 +17,33 @@
package org.apache.eventmesh.dashboard.core.function.SDK.operation;
+import static
org.apache.eventmesh.dashboard.core.function.SDK.util.RuntimeSDKOperationUtils.buildEventMeshGrpcProducerConfig;
+
+import org.apache.eventmesh.client.grpc.config.EventMeshGrpcClientConfig;
+import org.apache.eventmesh.client.grpc.producer.EventMeshGrpcProducer;
+import org.apache.eventmesh.common.exception.EventMeshException;
import org.apache.eventmesh.dashboard.core.function.SDK.AbstractSDKOperation;
-import
org.apache.eventmesh.dashboard.core.function.SDK.config.CreateEtcdConfig;
+import
org.apache.eventmesh.dashboard.core.function.SDK.config.CreateRuntimeConfig;
import org.apache.eventmesh.dashboard.core.function.SDK.config.CreateSDKConfig;
import java.util.AbstractMap.SimpleEntry;
-import io.etcd.jetcd.Client;
-import io.etcd.jetcd.KV;
-import io.etcd.jetcd.common.exception.EtcdException;
-
import lombok.extern.slf4j.Slf4j;
@Slf4j
-public class EtcdSDKOperation extends AbstractSDKOperation<KV> {
+public class RuntimeGrpcProducerSDKOperation extends
AbstractSDKOperation<EventMeshGrpcProducer> {
@Override
- public SimpleEntry<String, KV> createClient(CreateSDKConfig clientConfig) {
- final CreateEtcdConfig etcdConfig = (CreateEtcdConfig) clientConfig;
- KV kvClient = null;
+ public SimpleEntry<String, EventMeshGrpcProducer>
createClient(CreateSDKConfig clientConfig) {
+ final CreateRuntimeConfig runtimeConfig = (CreateRuntimeConfig)
clientConfig;
+ final EventMeshGrpcClientConfig grpcClientConfig =
buildEventMeshGrpcProducerConfig(runtimeConfig);
+ EventMeshGrpcProducer grpcProducer = null;
try {
- final Client client = Client.builder()
- .endpoints(getSplitEndpoints(etcdConfig))
- .build();
- kvClient = client.getKVClient();
- } catch (EtcdException e) {
- log.error("create etcd client failed", e);
+ grpcProducer = new EventMeshGrpcProducer(grpcClientConfig);
+ } catch (EventMeshException e) {
+ log.error("create runtime grpc Producer client failed", e);
}
- return new SimpleEntry<>(clientConfig.getUniqueKey(), kvClient);
- }
-
- private static String[] getSplitEndpoints(CreateEtcdConfig etcdConfig) {
- return etcdConfig.getEtcdServerAddress().split(",");
+ return new SimpleEntry<>(clientConfig.getUniqueKey(), grpcProducer);
}
@Override
diff --git
a/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/EtcdSDKOperation.java
b/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeHttpConsumerSDKOperation.java
similarity index 55%
copy from
eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/EtcdSDKOperation.java
copy to
eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeHttpConsumerSDKOperation.java
index b630a38..e6c026a 100644
---
a/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/EtcdSDKOperation.java
+++
b/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeHttpConsumerSDKOperation.java
@@ -17,38 +17,33 @@
package org.apache.eventmesh.dashboard.core.function.SDK.operation;
+import static
org.apache.eventmesh.dashboard.core.function.SDK.util.RuntimeSDKOperationUtils.buildEventMeshHttpConsumerConfig;
+
+import org.apache.eventmesh.client.http.conf.EventMeshHttpClientConfig;
+import org.apache.eventmesh.client.http.consumer.EventMeshHttpConsumer;
+import org.apache.eventmesh.common.exception.EventMeshException;
import org.apache.eventmesh.dashboard.core.function.SDK.AbstractSDKOperation;
-import
org.apache.eventmesh.dashboard.core.function.SDK.config.CreateEtcdConfig;
+import
org.apache.eventmesh.dashboard.core.function.SDK.config.CreateRuntimeConfig;
import org.apache.eventmesh.dashboard.core.function.SDK.config.CreateSDKConfig;
import java.util.AbstractMap.SimpleEntry;
-import io.etcd.jetcd.Client;
-import io.etcd.jetcd.KV;
-import io.etcd.jetcd.common.exception.EtcdException;
-
import lombok.extern.slf4j.Slf4j;
@Slf4j
-public class EtcdSDKOperation extends AbstractSDKOperation<KV> {
+public class RuntimeHttpConsumerSDKOperation extends
AbstractSDKOperation<EventMeshHttpConsumer> {
@Override
- public SimpleEntry<String, KV> createClient(CreateSDKConfig clientConfig) {
- final CreateEtcdConfig etcdConfig = (CreateEtcdConfig) clientConfig;
- KV kvClient = null;
+ public SimpleEntry<String, EventMeshHttpConsumer>
createClient(CreateSDKConfig clientConfig) {
+ final CreateRuntimeConfig runtimeConfig = (CreateRuntimeConfig)
clientConfig;
+ final EventMeshHttpClientConfig httpClientConfig =
buildEventMeshHttpConsumerConfig(runtimeConfig);
+ EventMeshHttpConsumer httpConsumer = null;
try {
- final Client client = Client.builder()
- .endpoints(getSplitEndpoints(etcdConfig))
- .build();
- kvClient = client.getKVClient();
- } catch (EtcdException e) {
- log.error("create etcd client failed", e);
+ httpConsumer = new EventMeshHttpConsumer(httpClientConfig);
+ } catch (EventMeshException e) {
+ log.error("create runtime http Consumer client failed", e);
}
- return new SimpleEntry<>(clientConfig.getUniqueKey(), kvClient);
- }
-
- private static String[] getSplitEndpoints(CreateEtcdConfig etcdConfig) {
- return etcdConfig.getEtcdServerAddress().split(",");
+ return new SimpleEntry<>(clientConfig.getUniqueKey(), httpConsumer);
}
@Override
diff --git
a/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/EtcdSDKOperation.java
b/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeHttpProducerSDKOperation.java
similarity index 55%
copy from
eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/EtcdSDKOperation.java
copy to
eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeHttpProducerSDKOperation.java
index b630a38..79071c7 100644
---
a/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/EtcdSDKOperation.java
+++
b/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeHttpProducerSDKOperation.java
@@ -17,38 +17,33 @@
package org.apache.eventmesh.dashboard.core.function.SDK.operation;
+import static
org.apache.eventmesh.dashboard.core.function.SDK.util.RuntimeSDKOperationUtils.buildEventMeshHttpProducerConfig;
+
+import org.apache.eventmesh.client.http.conf.EventMeshHttpClientConfig;
+import org.apache.eventmesh.client.http.producer.EventMeshHttpProducer;
+import org.apache.eventmesh.common.exception.EventMeshException;
import org.apache.eventmesh.dashboard.core.function.SDK.AbstractSDKOperation;
-import
org.apache.eventmesh.dashboard.core.function.SDK.config.CreateEtcdConfig;
+import
org.apache.eventmesh.dashboard.core.function.SDK.config.CreateRuntimeConfig;
import org.apache.eventmesh.dashboard.core.function.SDK.config.CreateSDKConfig;
import java.util.AbstractMap.SimpleEntry;
-import io.etcd.jetcd.Client;
-import io.etcd.jetcd.KV;
-import io.etcd.jetcd.common.exception.EtcdException;
-
import lombok.extern.slf4j.Slf4j;
@Slf4j
-public class EtcdSDKOperation extends AbstractSDKOperation<KV> {
+public class RuntimeHttpProducerSDKOperation extends
AbstractSDKOperation<EventMeshHttpProducer> {
@Override
- public SimpleEntry<String, KV> createClient(CreateSDKConfig clientConfig) {
- final CreateEtcdConfig etcdConfig = (CreateEtcdConfig) clientConfig;
- KV kvClient = null;
+ public SimpleEntry<String, EventMeshHttpProducer>
createClient(CreateSDKConfig clientConfig) {
+ final CreateRuntimeConfig runtimeConfig = (CreateRuntimeConfig)
clientConfig;
+ final EventMeshHttpClientConfig httpClientConfig =
buildEventMeshHttpProducerConfig(runtimeConfig);
+ EventMeshHttpProducer httpProducer = null;
try {
- final Client client = Client.builder()
- .endpoints(getSplitEndpoints(etcdConfig))
- .build();
- kvClient = client.getKVClient();
- } catch (EtcdException e) {
- log.error("create etcd client failed", e);
+ httpProducer = new EventMeshHttpProducer(httpClientConfig);
+ } catch (EventMeshException e) {
+ log.error("create runtime http Producer client failed", e);
}
- return new SimpleEntry<>(clientConfig.getUniqueKey(), kvClient);
- }
-
- private static String[] getSplitEndpoints(CreateEtcdConfig etcdConfig) {
- return etcdConfig.getEtcdServerAddress().split(",");
+ return new SimpleEntry<>(clientConfig.getUniqueKey(), httpProducer);
}
@Override
diff --git
a/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeSDKOperation.java
b/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeSDKOperation.java
index d630bc0..8ef5b81 100644
---
a/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeSDKOperation.java
+++
b/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeSDKOperation.java
@@ -17,19 +17,111 @@
package org.apache.eventmesh.dashboard.core.function.SDK.operation;
+import org.apache.eventmesh.client.grpc.consumer.EventMeshGrpcConsumer;
+import org.apache.eventmesh.client.grpc.producer.EventMeshGrpcProducer;
+import org.apache.eventmesh.client.http.consumer.EventMeshHttpConsumer;
+import org.apache.eventmesh.client.http.producer.EventMeshHttpProducer;
+import org.apache.eventmesh.client.tcp.impl.cloudevent.CloudEventTCPClient;
+import
org.apache.eventmesh.client.tcp.impl.eventmeshmessage.EventMeshMessageTCPClient;
+import org.apache.eventmesh.client.tcp.impl.openmessage.OpenMessageTCPClient;
+import org.apache.eventmesh.common.Constants;
import org.apache.eventmesh.dashboard.core.function.SDK.AbstractSDKOperation;
+import
org.apache.eventmesh.dashboard.core.function.SDK.config.CreateRuntimeConfig;
import org.apache.eventmesh.dashboard.core.function.SDK.config.CreateSDKConfig;
+import
org.apache.eventmesh.dashboard.core.function.SDK.wrapper.RuntimeSDKWrapper;
import java.util.AbstractMap.SimpleEntry;
-public class RuntimeSDKOperation extends AbstractSDKOperation {
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class RuntimeSDKOperation extends
AbstractSDKOperation<RuntimeSDKWrapper> {
+
+ private final RuntimeTcpCloudEventSDKOperation tcpCloudEventSDKOperation =
new RuntimeTcpCloudEventSDKOperation();
+ private final RuntimeTcpEventMeshSDKOperation tcpEventMeshSDKOperation =
new RuntimeTcpEventMeshSDKOperation();
+ private final RuntimeTcpOpenMessageSDKOperation tcpOpenMessageSDKOperation
= new RuntimeTcpOpenMessageSDKOperation();
+
+ private final RuntimeHttpProducerSDKOperation httpProducerSDKOperation =
new RuntimeHttpProducerSDKOperation();
+ private final RuntimeHttpConsumerSDKOperation httpConsumerSDKOperation =
new RuntimeHttpConsumerSDKOperation();
+
+ private final RuntimeGrpcProducerSDKOperation grpcProducerSDKOperation =
new RuntimeGrpcProducerSDKOperation();
+ private final RuntimeGrpcConsumerSDKOperation grpcConsumerSDKOperation =
new RuntimeGrpcConsumerSDKOperation();
+
@Override
- public SimpleEntry createClient(CreateSDKConfig clientConfig) {
- return null;
+ public SimpleEntry<String, RuntimeSDKWrapper> createClient(CreateSDKConfig
clientConfig) {
+ CreateRuntimeConfig runtimeConfig = (CreateRuntimeConfig) clientConfig;
+ final String protocol = ((CreateRuntimeConfig)
clientConfig).getProtocol();
+ final String protocolName = ((CreateRuntimeConfig)
clientConfig).getProtocolName();
+ final String clientType = ((CreateRuntimeConfig)
clientConfig).getClientType();
+
+ SimpleEntry<String, CloudEventTCPClient> cloudSimpleEntry = null;
+ SimpleEntry<String, EventMeshMessageTCPClient>
eventMeshMessageSimpleEntry = null;
+ SimpleEntry<String, OpenMessageTCPClient> openMessageSimpleEntry =
null;
+
+ SimpleEntry<String, EventMeshHttpProducer> httpProducerSimpleEntry =
null;
+ SimpleEntry<String, EventMeshHttpConsumer> httpConsumerSimpleEntry =
null;
+
+ SimpleEntry<String, EventMeshGrpcProducer> grpcProducerSimpleEntry =
null;
+ SimpleEntry<String, EventMeshGrpcConsumer> grpcConsumerSimpleEntry =
null;
+
+ switch (protocol) {
+ case Constants.TCP:
+ switch (protocolName) {
+ case Constants.CLOUD_EVENTS_PROTOCOL_NAME:
+ cloudSimpleEntry =
tcpCloudEventSDKOperation.createClient(runtimeConfig);
+ break;
+ case Constants.EM_MESSAGE_PROTOCOL_NAME:
+ eventMeshMessageSimpleEntry =
tcpEventMeshSDKOperation.createClient(runtimeConfig);
+ break;
+ case Constants.OPEN_MESSAGE_PROTOCOL_NAME:
+ openMessageSimpleEntry =
tcpOpenMessageSDKOperation.createClient(runtimeConfig);
+ break;
+ default:
+ break;
+ }
+ break;
+ case Constants.HTTP:
+ switch (clientType) {
+ case "producer":
+ httpProducerSimpleEntry =
httpProducerSDKOperation.createClient(runtimeConfig);
+ break;
+ case "consumer":
+ httpConsumerSimpleEntry =
httpConsumerSDKOperation.createClient(runtimeConfig);
+ break;
+ default:
+ break;
+ }
+ break;
+ case Constants.GRPC:
+ switch (clientType) {
+ case "producer":
+ grpcProducerSimpleEntry =
grpcProducerSDKOperation.createClient(runtimeConfig);
+ break;
+ case "consumer":
+ grpcConsumerSimpleEntry =
grpcConsumerSDKOperation.createClient(runtimeConfig);
+ break;
+ default:
+ break;
+ }
+ break;
+ default:
+ log.warn("clients that do not support the current protocol");
+ break;
+ }
+ RuntimeSDKWrapper runtimeClient = new RuntimeSDKWrapper(
+ cloudSimpleEntry != null ? cloudSimpleEntry.getValue() : null,
+ eventMeshMessageSimpleEntry != null ?
eventMeshMessageSimpleEntry.getValue() : null,
+ openMessageSimpleEntry != null ? openMessageSimpleEntry.getValue()
: null,
+ httpProducerSimpleEntry != null ?
httpProducerSimpleEntry.getValue() : null,
+ httpConsumerSimpleEntry != null ?
httpConsumerSimpleEntry.getValue() : null,
+ grpcProducerSimpleEntry != null ?
grpcProducerSimpleEntry.getValue() : null,
+ grpcConsumerSimpleEntry != null ?
grpcConsumerSimpleEntry.getValue() : null
+ );
+ return new SimpleEntry<>(clientConfig.getUniqueKey(), runtimeClient);
}
@Override
public void close(Object client) {
-
+ castClient(client).close();
}
}
diff --git
a/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeTcpCloudEventSDKOperation.java
b/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeTcpCloudEventSDKOperation.java
new file mode 100644
index 0000000..e097e2c
--- /dev/null
+++
b/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeTcpCloudEventSDKOperation.java
@@ -0,0 +1,62 @@
+/*
+ * 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.dashboard.core.function.SDK.operation;
+
+import static
org.apache.eventmesh.dashboard.core.function.SDK.util.RuntimeSDKOperationUtils.buildEventMeshTCPClientConfig;
+import static
org.apache.eventmesh.dashboard.core.function.SDK.util.RuntimeSDKOperationUtils.buildUserAgent;
+
+import org.apache.eventmesh.client.tcp.conf.EventMeshTCPClientConfig;
+import org.apache.eventmesh.client.tcp.impl.cloudevent.CloudEventTCPClient;
+import org.apache.eventmesh.common.exception.EventMeshException;
+import org.apache.eventmesh.common.protocol.tcp.UserAgent;
+import org.apache.eventmesh.dashboard.core.function.SDK.AbstractSDKOperation;
+import
org.apache.eventmesh.dashboard.core.function.SDK.config.CreateRuntimeConfig;
+import org.apache.eventmesh.dashboard.core.function.SDK.config.CreateSDKConfig;
+
+import java.util.AbstractMap.SimpleEntry;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class RuntimeTcpCloudEventSDKOperation extends
AbstractSDKOperation<CloudEventTCPClient> {
+
+ @Override
+ public SimpleEntry<String, CloudEventTCPClient>
createClient(CreateSDKConfig clientConfig) {
+ final CreateRuntimeConfig runtimeConfig = (CreateRuntimeConfig)
clientConfig;
+ CloudEventTCPClient cloudEventTCPClient = null;
+ try {
+ final UserAgent userAgent =
buildUserAgent(runtimeConfig.getUserAgent());
+ final EventMeshTCPClientConfig eventMeshTCPClientConfig =
buildEventMeshTCPClientConfig(
+ runtimeConfig.getRuntimeServerAddress(), userAgent);
+ cloudEventTCPClient = new
CloudEventTCPClient(eventMeshTCPClientConfig);
+ cloudEventTCPClient.init();
+ } catch (EventMeshException e) {
+ log.error("create runtime CloudEvent tcp client failed", e);
+ }
+ return new SimpleEntry<>(clientConfig.getUniqueKey(),
cloudEventTCPClient);
+ }
+
+ @Override
+ public void close(Object client) {
+ try {
+ castClient(client).close();
+ } catch (Exception e) {
+ log.error("close eventmesh runtime tcp client failed");
+ }
+ }
+}
diff --git
a/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeTcpEventMeshSDKOperation.java
b/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeTcpEventMeshSDKOperation.java
new file mode 100644
index 0000000..45f7152
--- /dev/null
+++
b/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeTcpEventMeshSDKOperation.java
@@ -0,0 +1,62 @@
+/*
+ * 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.dashboard.core.function.SDK.operation;
+
+import static
org.apache.eventmesh.dashboard.core.function.SDK.util.RuntimeSDKOperationUtils.buildEventMeshTCPClientConfig;
+import static
org.apache.eventmesh.dashboard.core.function.SDK.util.RuntimeSDKOperationUtils.buildUserAgent;
+
+import org.apache.eventmesh.client.tcp.conf.EventMeshTCPClientConfig;
+import
org.apache.eventmesh.client.tcp.impl.eventmeshmessage.EventMeshMessageTCPClient;
+import org.apache.eventmesh.common.exception.EventMeshException;
+import org.apache.eventmesh.common.protocol.tcp.UserAgent;
+import org.apache.eventmesh.dashboard.core.function.SDK.AbstractSDKOperation;
+import
org.apache.eventmesh.dashboard.core.function.SDK.config.CreateRuntimeConfig;
+import org.apache.eventmesh.dashboard.core.function.SDK.config.CreateSDKConfig;
+
+import java.util.AbstractMap.SimpleEntry;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class RuntimeTcpEventMeshSDKOperation extends
AbstractSDKOperation<EventMeshMessageTCPClient> {
+
+ @Override
+ public SimpleEntry<String, EventMeshMessageTCPClient>
createClient(CreateSDKConfig clientConfig) {
+ final CreateRuntimeConfig runtimeConfig = (CreateRuntimeConfig)
clientConfig;
+ EventMeshMessageTCPClient eventMeshTCPClient = null;
+ try {
+ UserAgent userAgent = buildUserAgent(runtimeConfig.getUserAgent());
+ final EventMeshTCPClientConfig eventMeshTCPClientConfig =
buildEventMeshTCPClientConfig(
+ runtimeConfig.getRuntimeServerAddress(), userAgent);
+ eventMeshTCPClient = new
EventMeshMessageTCPClient(eventMeshTCPClientConfig);
+ eventMeshTCPClient.init();
+ } catch (EventMeshException e) {
+ log.error("create runtime EventMeshMessage tcp client failed", e);
+ }
+ return new SimpleEntry<>(clientConfig.getUniqueKey(),
eventMeshTCPClient);
+ }
+
+ @Override
+ public void close(Object client) {
+ try {
+ castClient(client).close();
+ } catch (Exception e) {
+ log.error("EventMeshMessage client close failed", e);
+ }
+ }
+}
diff --git
a/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeTcpOpenMessageSDKOperation.java
b/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeTcpOpenMessageSDKOperation.java
new file mode 100644
index 0000000..0a28106
--- /dev/null
+++
b/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeTcpOpenMessageSDKOperation.java
@@ -0,0 +1,57 @@
+/*
+ * 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.dashboard.core.function.SDK.operation;
+
+import static
org.apache.eventmesh.dashboard.core.function.SDK.util.RuntimeSDKOperationUtils.buildEventMeshTCPClientConfig;
+import static
org.apache.eventmesh.dashboard.core.function.SDK.util.RuntimeSDKOperationUtils.buildUserAgent;
+
+import org.apache.eventmesh.client.tcp.conf.EventMeshTCPClientConfig;
+import org.apache.eventmesh.client.tcp.impl.openmessage.OpenMessageTCPClient;
+import org.apache.eventmesh.common.protocol.tcp.UserAgent;
+import org.apache.eventmesh.dashboard.core.function.SDK.AbstractSDKOperation;
+import
org.apache.eventmesh.dashboard.core.function.SDK.config.CreateRuntimeConfig;
+import org.apache.eventmesh.dashboard.core.function.SDK.config.CreateSDKConfig;
+
+import java.util.AbstractMap;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class RuntimeTcpOpenMessageSDKOperation extends
AbstractSDKOperation<OpenMessageTCPClient> {
+
+ @Override
+ public AbstractMap.SimpleEntry<String, OpenMessageTCPClient>
createClient(CreateSDKConfig clientConfig) {
+ final CreateRuntimeConfig runtimeConfig = (CreateRuntimeConfig)
clientConfig;
+ OpenMessageTCPClient openMessageTCPClient = null;
+ try {
+ UserAgent userAgent = buildUserAgent(runtimeConfig.getUserAgent());
+ final EventMeshTCPClientConfig eventMeshTCPClientConfig =
buildEventMeshTCPClientConfig(
+ runtimeConfig.getRuntimeServerAddress(), userAgent);
+ openMessageTCPClient = new
OpenMessageTCPClient(eventMeshTCPClientConfig);
+ openMessageTCPClient.init();
+ } catch (Exception e) {
+ log.error("create runtime eventmesh OpenMessage client failed", e);
+ }
+ return new AbstractMap.SimpleEntry<>(clientConfig.getUniqueKey(),
openMessageTCPClient);
+ }
+
+ @Override
+ public void close(Object client) {
+ castClient(client).close();
+ }
+}
diff --git
a/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/util/RuntimeSDKOperationUtils.java
b/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/util/RuntimeSDKOperationUtils.java
new file mode 100644
index 0000000..435949c
--- /dev/null
+++
b/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/util/RuntimeSDKOperationUtils.java
@@ -0,0 +1,107 @@
+/*
+ * 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.dashboard.core.function.SDK.util;
+
+import org.apache.eventmesh.client.grpc.config.EventMeshGrpcClientConfig;
+import org.apache.eventmesh.client.http.conf.EventMeshHttpClientConfig;
+import org.apache.eventmesh.client.tcp.conf.EventMeshTCPClientConfig;
+import org.apache.eventmesh.common.protocol.tcp.UserAgent;
+import
org.apache.eventmesh.dashboard.core.function.SDK.config.CreateRuntimeConfig;
+
+public class RuntimeSDKOperationUtils {
+
+ public static EventMeshTCPClientConfig
buildEventMeshTCPClientConfig(String serverAddress, UserAgent userAgent) {
+ String clientHost = serverAddress.split(":")[0];
+ int clientPort = Integer.parseInt(serverAddress.split(":")[1]);
+ return EventMeshTCPClientConfig.builder()
+ .host(clientHost)
+ .port(clientPort)
+ .userAgent(userAgent)
+ .build();
+ }
+
+ public static UserAgent buildUserAgent(UserAgent userAgent) {
+ return UserAgent.builder()
+ .host(userAgent.getHost())
+ .subsystem(userAgent.getSubsystem())
+ .port(userAgent.getPort())
+ .group(userAgent.getGroup())
+ .idc(userAgent.getIdc())
+ .env(userAgent.getPassword())
+ .pid(userAgent.getPid())
+ .path(userAgent.getPath())
+ .username(userAgent.getUsername())
+ .password(userAgent.getPassword())
+ .version(userAgent.getVersion())
+ .purpose(userAgent.getPurpose())
+ .build();
+ }
+
+ public static EventMeshHttpClientConfig
buildEventMeshHttpConsumerConfig(CreateRuntimeConfig runtimeConfig) {
+ return EventMeshHttpClientConfig.builder()
+ .liteEventMeshAddr(runtimeConfig.getRuntimeServerAddress())
+ .consumerGroup(runtimeConfig.getConsumerGroup())
+ .sys(runtimeConfig.getSys())
+ .env(runtimeConfig.getEnv())
+ .idc(runtimeConfig.getIdc())
+ .ip(runtimeConfig.getIp())
+ .pid(runtimeConfig.getPid())
+ .build();
+ }
+
+ public static EventMeshHttpClientConfig
buildEventMeshHttpProducerConfig(CreateRuntimeConfig runtimeConfig) {
+ return EventMeshHttpClientConfig.builder()
+ .liteEventMeshAddr(runtimeConfig.getRuntimeServerAddress())
+ .producerGroup(runtimeConfig.getProducerGroup())
+ .env(runtimeConfig.getEnv())
+ .idc(runtimeConfig.getIdc())
+ .sys(runtimeConfig.getSys())
+ .pid(runtimeConfig.getPid())
+ .userName(runtimeConfig.getUsername())
+ .password(runtimeConfig.getPassword())
+ .build();
+ }
+
+ public static EventMeshGrpcClientConfig
buildEventMeshGrpcConsumerConfig(CreateRuntimeConfig runtimeConfig) {
+ final String grpcServerAddress =
runtimeConfig.getRuntimeServerAddress();
+ String clientHost = grpcServerAddress.split(":")[0];
+ int clientPort = Integer.parseInt(grpcServerAddress.split(":")[1]);
+ return EventMeshGrpcClientConfig.builder()
+ .serverAddr(clientHost)
+ .serverPort(clientPort)
+ .consumerGroup(runtimeConfig.getConsumerGroup())
+ .env(runtimeConfig.getEnv())
+ .idc(runtimeConfig.getIdc())
+ .sys(runtimeConfig.getSys())
+ .build();
+ }
+
+ public static EventMeshGrpcClientConfig
buildEventMeshGrpcProducerConfig(CreateRuntimeConfig runtimeConfig) {
+ final String grpcServerAddress =
runtimeConfig.getRuntimeServerAddress();
+ String clientHost = grpcServerAddress.split(":")[0];
+ int clientPort = Integer.parseInt(grpcServerAddress.split(":")[1]);
+ return EventMeshGrpcClientConfig.builder()
+ .serverAddr(clientHost)
+ .serverPort(clientPort)
+ .producerGroup(runtimeConfig.getProducerGroup())
+ .env(runtimeConfig.getEnv())
+ .idc(runtimeConfig.getIdc())
+ .sys(runtimeConfig.getSys())
+ .build();
+ }
+}
diff --git
a/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/wrapper/NacosSDKWrapper.java
b/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/wrapper/NacosSDKWrapper.java
index a0c0858..60f9022 100644
---
a/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/wrapper/NacosSDKWrapper.java
+++
b/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/wrapper/NacosSDKWrapper.java
@@ -22,7 +22,9 @@ import com.alibaba.nacos.api.exception.NacosException;
import com.alibaba.nacos.api.naming.NamingService;
import lombok.AllArgsConstructor;
+import lombok.Data;
+@Data
@AllArgsConstructor
public class NacosSDKWrapper {
diff --git
a/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/wrapper/RuntimeSDKWrapper.java
b/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/wrapper/RuntimeSDKWrapper.java
new file mode 100644
index 0000000..9e6370b
--- /dev/null
+++
b/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/wrapper/RuntimeSDKWrapper.java
@@ -0,0 +1,71 @@
+/*
+ * 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.dashboard.core.function.SDK.wrapper;
+
+import org.apache.eventmesh.client.grpc.consumer.EventMeshGrpcConsumer;
+import org.apache.eventmesh.client.grpc.producer.EventMeshGrpcProducer;
+import org.apache.eventmesh.client.http.consumer.EventMeshHttpConsumer;
+import org.apache.eventmesh.client.http.producer.EventMeshHttpProducer;
+import org.apache.eventmesh.client.tcp.impl.cloudevent.CloudEventTCPClient;
+import
org.apache.eventmesh.client.tcp.impl.eventmeshmessage.EventMeshMessageTCPClient;
+import org.apache.eventmesh.client.tcp.impl.openmessage.OpenMessageTCPClient;
+import org.apache.eventmesh.common.exception.EventMeshException;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+@Slf4j
+public class RuntimeSDKWrapper {
+
+ private CloudEventTCPClient tcpCloudEventClient;
+ private EventMeshMessageTCPClient tcpEventMeshClient;
+ private OpenMessageTCPClient openMessageTCPClient;
+
+ private EventMeshHttpProducer httpProducerClient;
+ private EventMeshHttpConsumer httpConsumerClient;
+
+ private EventMeshGrpcProducer grpcProducerClient;
+ private EventMeshGrpcConsumer grpcConsumerClient;
+
+ public void close() {
+ try {
+ if (tcpCloudEventClient != null) {
+ tcpCloudEventClient.close();
+ } else if (tcpEventMeshClient != null) {
+ tcpEventMeshClient.close();
+ } else if (openMessageTCPClient != null) {
+ openMessageTCPClient.close();
+ } else if (httpProducerClient != null) {
+ httpProducerClient.close();
+ } else if (httpConsumerClient != null) {
+ httpConsumerClient.close();
+ } else if (grpcProducerClient != null) {
+ grpcProducerClient.close();
+ } else if (grpcConsumerClient != null) {
+ grpcConsumerClient.close();
+ }
+ } catch (EventMeshException e) {
+ log.error("runtime client close failed", e);
+ }
+ }
+}
diff --git
a/eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/SDKManagerTest.java
b/eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/SDKManagerTest.java
index 7d9008f..e936696 100644
---
a/eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/SDKManagerTest.java
+++
b/eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/SDKManagerTest.java
@@ -30,13 +30,17 @@ import lombok.extern.slf4j.Slf4j;
@Slf4j
class SDKManagerTest {
- private final CreateRedisConfig createRedisConfig = new
CreateRedisConfig();
private String redisKey;
@BeforeEach
void setUp() {
try {
- createRedisConfig.setRedisUrl("redis://localhost:6379");
+ CreateRedisConfig createRedisConfig = CreateRedisConfig.builder()
+ .redisUrl("localhost:6379")
+ .password("")
+ .timeOut(30)
+ .build();
+ // createRedisConfig.setRedisUrl("redis://localhost:6379");
redisKey =
SDKManager.getInstance().createClient(SDKTypeEnum.STORAGE_REDIS,
createRedisConfig).getKey();
} catch (Exception e) {
log.warn("SDK manager test init failed, possible reason:
redis-server is offline. {}", this.getClass().getSimpleName(), e);
diff --git
a/eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/EtcdSDKCreateOperationTest.java
b/eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/EtcdSDKCreateOperationTest.java
index a4d8f4b..13afca1 100644
---
a/eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/EtcdSDKCreateOperationTest.java
+++
b/eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/EtcdSDKCreateOperationTest.java
@@ -19,17 +19,12 @@ package
org.apache.eventmesh.dashboard.core.function.SDK.operation;
import
org.apache.eventmesh.dashboard.core.function.SDK.config.CreateEtcdConfig;
-import java.nio.charset.StandardCharsets;
import java.util.AbstractMap.SimpleEntry;
-import java.util.List;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
-import io.etcd.jetcd.ByteSequence;
import io.etcd.jetcd.KV;
-import io.etcd.jetcd.KeyValue;
-import io.etcd.jetcd.kv.GetResponse;
import lombok.extern.slf4j.Slf4j;
@@ -38,31 +33,24 @@ public class EtcdSDKCreateOperationTest {
private final EtcdSDKOperation etcdSDKOperation = new EtcdSDKOperation();
- private static final String key = "/test/foo";
-
- private static final String value = "test";
-
- private static final String url = "http://127.0.0.1:2379";;
+ private static final String url = "http://localhost:2379";;
@Test
void testCreateClient() {
- final CreateEtcdConfig etcdConfig = new CreateEtcdConfig();
- etcdConfig.setEtcdServerAddress(url);
+ final CreateEtcdConfig etcdConfig = CreateEtcdConfig.builder()
+ .etcdServerAddress(url)
+ .connectTime(5)
+ .build();
+ SimpleEntry<String, KV> simpleEntry = null;
try {
- final SimpleEntry<String, KV> simpleEntry =
etcdSDKOperation.createClient(etcdConfig);
+ simpleEntry = etcdSDKOperation.createClient(etcdConfig);
Assertions.assertEquals(url, simpleEntry.getKey());
- simpleEntry.getValue().put(bytesOf(key), bytesOf(value));
- final GetResponse response =
simpleEntry.getValue().get(bytesOf(key)).get();
- final List<KeyValue> keyValues = response.getKvs();
- log.info("get key = {} , value = {} from etcd success",
- keyValues.get(0).getKey().toString(StandardCharsets.UTF_8),
- keyValues.get(0).getValue().toString(StandardCharsets.UTF_8));
+ simpleEntry.getValue().close();
} catch (Exception e) {
log.error("create etcd client failed", e);
+ if (simpleEntry != null) {
+ simpleEntry.getValue().close();
+ }
}
}
-
- private static ByteSequence bytesOf(String val) {
- return ByteSequence.from(val, StandardCharsets.UTF_8);
- }
}
diff --git
a/eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RedisSDKCreateOperationTest.java
b/eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RedisSDKCreateOperationTest.java
index ec90f29..2129d74 100644
---
a/eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RedisSDKCreateOperationTest.java
+++
b/eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RedisSDKCreateOperationTest.java
@@ -32,20 +32,27 @@ import lombok.extern.slf4j.Slf4j;
@Slf4j
class RedisSDKCreateOperationTest {
- private RedisSDKOperation redisClientCreateOperation = new
RedisSDKOperation();
+ private final RedisSDKOperation redisClientCreateOperation = new
RedisSDKOperation();
@Test
void testCreateClient() {
- CreateRedisConfig createClientConfig = new CreateRedisConfig();
- createClientConfig.setRedisUrl("redis://localhost:6379");
+ CreateRedisConfig createClientConfig = CreateRedisConfig.builder()
+ .redisUrl("localhost:6379")
+ .password("")
+ .timeOut(5)
+ .build();
+ SimpleEntry<String, StatefulRedisConnection<String, String>>
simpleEntry = null;
try {
- SimpleEntry<String, StatefulRedisConnection<String, String>>
simpleEntry = redisClientCreateOperation.createClient(createClientConfig);
- assertEquals("redis://localhost:6379", simpleEntry.getKey());
+ simpleEntry =
redisClientCreateOperation.createClient(createClientConfig);
+ assertEquals("localhost:6379", simpleEntry.getKey());
String response = simpleEntry.getValue().sync().ping();
log.info("response:{}", response);
+ simpleEntry.getValue().close();
} catch (Exception e) {
log.error("create redis client failed", e);
+ if (simpleEntry != null) {
+ simpleEntry.getValue().close();
+ }
}
-
}
}
\ No newline at end of file
diff --git
a/eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RedisSDKCreateOperationTest.java
b/eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeGrpcConsumerSDKOperationTest.java
similarity index 50%
copy from
eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RedisSDKCreateOperationTest.java
copy to
eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeGrpcConsumerSDKOperationTest.java
index ec90f29..a3869a6 100644
---
a/eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RedisSDKCreateOperationTest.java
+++
b/eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeGrpcConsumerSDKOperationTest.java
@@ -17,35 +17,40 @@
package org.apache.eventmesh.dashboard.core.function.SDK.operation;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-
-import
org.apache.eventmesh.dashboard.core.function.SDK.config.CreateRedisConfig;
+import org.apache.eventmesh.client.grpc.consumer.EventMeshGrpcConsumer;
+import
org.apache.eventmesh.dashboard.core.function.SDK.config.CreateRuntimeConfig;
import java.util.AbstractMap.SimpleEntry;
+import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
-import io.lettuce.core.api.StatefulRedisConnection;
-
import lombok.extern.slf4j.Slf4j;
@Slf4j
-class RedisSDKCreateOperationTest {
+public class RuntimeGrpcConsumerSDKOperationTest {
- private RedisSDKOperation redisClientCreateOperation = new
RedisSDKOperation();
+ private final RuntimeGrpcConsumerSDKOperation grpcConsumerSDKOperation =
new RuntimeGrpcConsumerSDKOperation();
@Test
void testCreateClient() {
- CreateRedisConfig createClientConfig = new CreateRedisConfig();
- createClientConfig.setRedisUrl("redis://localhost:6379");
+ SimpleEntry<String, EventMeshGrpcConsumer> grpcConsumerSimpleEntry =
null;
try {
- SimpleEntry<String, StatefulRedisConnection<String, String>>
simpleEntry = redisClientCreateOperation.createClient(createClientConfig);
- assertEquals("redis://localhost:6379", simpleEntry.getKey());
- String response = simpleEntry.getValue().sync().ping();
- log.info("response:{}", response);
+ final CreateRuntimeConfig runtimeConfig =
CreateRuntimeConfig.builder()
+ .runtimeServerAddress("127.0.0.1:10205")
+ .consumerGroup("EventMeshTest-consumerGroup")
+ .env("test")
+ .idc("idc")
+ .sys("1234")
+ .build();
+ grpcConsumerSimpleEntry =
grpcConsumerSDKOperation.createClient(runtimeConfig);
+ Assertions.assertEquals("127.0.0.1:10205",
grpcConsumerSimpleEntry.getKey());
+ grpcConsumerSimpleEntry.getValue().close();
} catch (Exception e) {
- log.error("create redis client failed", e);
+ log.error("create runtime GRPC consumer client failed", e);
+ if (grpcConsumerSimpleEntry != null) {
+ grpcConsumerSimpleEntry.getValue().close();
+ }
}
-
}
-}
\ No newline at end of file
+}
diff --git
a/eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RedisSDKCreateOperationTest.java
b/eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeGrpcProducerSDKOperationTest.java
similarity index 50%
copy from
eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RedisSDKCreateOperationTest.java
copy to
eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeGrpcProducerSDKOperationTest.java
index ec90f29..f593f2f 100644
---
a/eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RedisSDKCreateOperationTest.java
+++
b/eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeGrpcProducerSDKOperationTest.java
@@ -17,35 +17,40 @@
package org.apache.eventmesh.dashboard.core.function.SDK.operation;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-
-import
org.apache.eventmesh.dashboard.core.function.SDK.config.CreateRedisConfig;
+import org.apache.eventmesh.client.grpc.producer.EventMeshGrpcProducer;
+import
org.apache.eventmesh.dashboard.core.function.SDK.config.CreateRuntimeConfig;
import java.util.AbstractMap.SimpleEntry;
+import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
-import io.lettuce.core.api.StatefulRedisConnection;
-
import lombok.extern.slf4j.Slf4j;
@Slf4j
-class RedisSDKCreateOperationTest {
+public class RuntimeGrpcProducerSDKOperationTest {
- private RedisSDKOperation redisClientCreateOperation = new
RedisSDKOperation();
+ private final RuntimeGrpcProducerSDKOperation grpcProducerSDKOperation =
new RuntimeGrpcProducerSDKOperation();
@Test
void testCreateClient() {
- CreateRedisConfig createClientConfig = new CreateRedisConfig();
- createClientConfig.setRedisUrl("redis://localhost:6379");
+ SimpleEntry<String, EventMeshGrpcProducer> grpcProducerSimpleEntry =
null;
try {
- SimpleEntry<String, StatefulRedisConnection<String, String>>
simpleEntry = redisClientCreateOperation.createClient(createClientConfig);
- assertEquals("redis://localhost:6379", simpleEntry.getKey());
- String response = simpleEntry.getValue().sync().ping();
- log.info("response:{}", response);
+ final CreateRuntimeConfig runtimeConfig =
CreateRuntimeConfig.builder()
+ .runtimeServerAddress("127.0.0.1:10205")
+ .producerGroup("EventMeshTest-producerGroup")
+ .env("test")
+ .idc("idc")
+ .sys("1234")
+ .build();
+ grpcProducerSimpleEntry =
grpcProducerSDKOperation.createClient(runtimeConfig);
+ Assertions.assertEquals("127.0.0.1:10205",
grpcProducerSimpleEntry.getKey());
+ grpcProducerSimpleEntry.getValue().close();
} catch (Exception e) {
- log.error("create redis client failed", e);
+ log.error("create runtime GRPC producer client failed", e);
+ if (grpcProducerSimpleEntry != null) {
+ grpcProducerSimpleEntry.getValue().close();
+ }
}
-
}
-}
\ No newline at end of file
+}
diff --git
a/eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeHttpConsumerSDKOperationTest.java
b/eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeHttpConsumerSDKOperationTest.java
new file mode 100644
index 0000000..0d12969
--- /dev/null
+++
b/eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeHttpConsumerSDKOperationTest.java
@@ -0,0 +1,62 @@
+/*
+ * 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.dashboard.core.function.SDK.operation;
+
+import org.apache.eventmesh.client.http.consumer.EventMeshHttpConsumer;
+import org.apache.eventmesh.common.utils.IPUtils;
+import org.apache.eventmesh.common.utils.ThreadUtils;
+import
org.apache.eventmesh.dashboard.core.function.SDK.config.CreateRuntimeConfig;
+
+import java.util.AbstractMap.SimpleEntry;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class RuntimeHttpConsumerSDKOperationTest {
+
+ private final RuntimeHttpConsumerSDKOperation httpConsumerSDKOperation =
new RuntimeHttpConsumerSDKOperation();
+
+ @Test
+ void testCreateClient() {
+ SimpleEntry<String, EventMeshHttpConsumer> httpConsumerSimpleEntry =
null;
+ try {
+ final CreateRuntimeConfig runtimeConfig =
CreateRuntimeConfig.builder()
+ .runtimeServerAddress("127.0.0.1:10105")
+ .consumerGroup("EventMeshTest-consumerGroup")
+ .env("test")
+ .idc("idc")
+ .ip(IPUtils.getLocalAddress())
+ .sys("1234")
+ .pid(String.valueOf(ThreadUtils.getPID()))
+ .username("eventmesh")
+ .password("123456")
+ .build();
+ httpConsumerSimpleEntry =
httpConsumerSDKOperation.createClient(runtimeConfig);
+ Assertions.assertEquals("127.0.0.1:10105",
httpConsumerSimpleEntry.getKey());
+ httpConsumerSimpleEntry.getValue().close();
+ } catch (Exception e) {
+ log.error("create runtime GRPC consumer client failed", e);
+ if (httpConsumerSimpleEntry != null) {
+ httpConsumerSimpleEntry.getValue().close();
+ }
+ }
+ }
+}
diff --git
a/eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeHttpProducerSDKOperationTest.java
b/eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeHttpProducerSDKOperationTest.java
new file mode 100644
index 0000000..1407214
--- /dev/null
+++
b/eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeHttpProducerSDKOperationTest.java
@@ -0,0 +1,62 @@
+/*
+ * 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.dashboard.core.function.SDK.operation;
+
+import org.apache.eventmesh.client.http.producer.EventMeshHttpProducer;
+import org.apache.eventmesh.common.utils.IPUtils;
+import org.apache.eventmesh.common.utils.ThreadUtils;
+import
org.apache.eventmesh.dashboard.core.function.SDK.config.CreateRuntimeConfig;
+
+import java.util.AbstractMap.SimpleEntry;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class RuntimeHttpProducerSDKOperationTest {
+
+ private final RuntimeHttpProducerSDKOperation httpProducerSDKOperation =
new RuntimeHttpProducerSDKOperation();
+
+ @Test
+ void testCreateClient() {
+ SimpleEntry<String, EventMeshHttpProducer> httpProducerSimpleEntry =
null;
+ try {
+ final CreateRuntimeConfig runtimeConfig =
CreateRuntimeConfig.builder()
+ .runtimeServerAddress("127.0.0.1:10105")
+ .producerGroup("EventMeshTest-producerGroup")
+ .env("test")
+ .idc("idc")
+ .ip(IPUtils.getLocalAddress())
+ .sys("1234")
+ .pid(String.valueOf(ThreadUtils.getPID()))
+ .username("eventmesh")
+ .password("123456")
+ .build();
+ httpProducerSimpleEntry =
httpProducerSDKOperation.createClient(runtimeConfig);
+ Assertions.assertEquals("127.0.0.1:10105",
httpProducerSimpleEntry.getKey());
+ httpProducerSimpleEntry.getValue().close();
+ } catch (Exception e) {
+ log.error("create runtime EventMesh HTTP producer client failed",
e);
+ if (httpProducerSimpleEntry != null) {
+ httpProducerSimpleEntry.getValue().close();
+ }
+ }
+ }
+}
diff --git
a/eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeSDKOperationTest.java
b/eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeSDKOperationTest.java
new file mode 100644
index 0000000..8f8d209
--- /dev/null
+++
b/eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeSDKOperationTest.java
@@ -0,0 +1,73 @@
+/*
+ * 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.dashboard.core.function.SDK.operation;
+
+import org.apache.eventmesh.client.tcp.common.EventMeshCommon;
+import org.apache.eventmesh.common.Constants;
+import org.apache.eventmesh.common.protocol.tcp.UserAgent;
+import
org.apache.eventmesh.dashboard.core.function.SDK.config.CreateRuntimeConfig;
+import
org.apache.eventmesh.dashboard.core.function.SDK.wrapper.RuntimeSDKWrapper;
+
+import java.util.AbstractMap.SimpleEntry;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+class RuntimeSDKOperationTest {
+
+ private final RuntimeSDKOperation runtimeSDKOperation = new
RuntimeSDKOperation();
+
+ @Test
+ void testCreateClient() {
+ SimpleEntry<String, RuntimeSDKWrapper> sdkWrapperSimpleEntry = null;
+ try {
+ final UserAgent userAgent = UserAgent.builder()
+ .env("test")
+ .host("localhost")
+ .password("123456")
+ .username("eventmesh")
+ .group("EventmeshTestGroup")
+ .path("/")
+ .port(8366)
+ .subsystem("502")
+ .pid(32894)
+ .version("2.1")
+ .idc("A")
+ .purpose(EventMeshCommon.USER_AGENT_PURPOSE_PUB)
+ .build();
+ final CreateRuntimeConfig runtimeConfig =
CreateRuntimeConfig.builder()
+ .runtimeServerAddress("127.0.0.1:10000")
+ .protocol("TCP")
+ .protocolName(Constants.EM_MESSAGE_PROTOCOL_NAME)
+ .userAgent(userAgent)
+ .build();
+ sdkWrapperSimpleEntry =
runtimeSDKOperation.createClient(runtimeConfig);
+ Assertions.assertEquals("127.0.0.1:10000",
sdkWrapperSimpleEntry.getKey());
+
Assertions.assertNotNull(sdkWrapperSimpleEntry.getValue().getTcpEventMeshClient());
+ sdkWrapperSimpleEntry.getValue().close();
+ } catch (Exception e) {
+ log.error("create runtime client failed", e);
+ if (sdkWrapperSimpleEntry != null) {
+ sdkWrapperSimpleEntry.getValue().close();
+ }
+ }
+ }
+}
\ No newline at end of file
diff --git
a/eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeTcpCloudEventSDKOperationTest.java
b/eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeTcpCloudEventSDKOperationTest.java
new file mode 100644
index 0000000..ca43b48
--- /dev/null
+++
b/eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeTcpCloudEventSDKOperationTest.java
@@ -0,0 +1,71 @@
+/*
+ * 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.dashboard.core.function.SDK.operation;
+
+import org.apache.eventmesh.client.tcp.common.EventMeshCommon;
+import org.apache.eventmesh.client.tcp.impl.cloudevent.CloudEventTCPClient;
+import org.apache.eventmesh.common.protocol.tcp.UserAgent;
+import
org.apache.eventmesh.dashboard.core.function.SDK.config.CreateRuntimeConfig;
+
+import java.util.AbstractMap.SimpleEntry;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class RuntimeTcpCloudEventSDKOperationTest {
+
+ private final RuntimeTcpCloudEventSDKOperation runtimeTCPPushSDKOperation
= new RuntimeTcpCloudEventSDKOperation();
+
+ @Test
+ void testCreateClient() {
+ SimpleEntry<String, CloudEventTCPClient> simpleEntry = null;
+ try {
+ final UserAgent userAgent = UserAgent.builder()
+ .env("test")
+ .host("localhost")
+ .password("123456")
+ .username("eventmesh")
+ .group("EventmeshTestGroup")
+ .path("/")
+ .port(8366)
+ .subsystem("502")
+ .pid(32894)
+ .version("2.1")
+ .idc("A")
+ .purpose(EventMeshCommon.USER_AGENT_PURPOSE_PUB)
+ .build();
+ log.info("{}", userAgent);
+ final CreateRuntimeConfig runtimeConfig =
CreateRuntimeConfig.builder()
+ .runtimeServerAddress("127.0.0.1:10000")
+ .userAgent(userAgent)
+ .build();
+ log.info("{}", runtimeConfig);
+ simpleEntry =
runtimeTCPPushSDKOperation.createClient(runtimeConfig);
+ Assertions.assertEquals("127.0.0.1:10000", simpleEntry.getKey());
+ simpleEntry.getValue().close();
+ } catch (Exception e) {
+ log.error("create runtime tcp CloudEvent client failed", e);
+ if (simpleEntry != null) {
+ simpleEntry.getValue().close();
+ }
+ }
+ }
+}
diff --git
a/eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeTcpEventMeshSDKOperationTest.java
b/eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeTcpEventMeshSDKOperationTest.java
new file mode 100644
index 0000000..733d39e
--- /dev/null
+++
b/eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeTcpEventMeshSDKOperationTest.java
@@ -0,0 +1,71 @@
+/*
+ * 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.dashboard.core.function.SDK.operation;
+
+import org.apache.eventmesh.client.tcp.common.EventMeshCommon;
+import
org.apache.eventmesh.client.tcp.impl.eventmeshmessage.EventMeshMessageTCPClient;
+import org.apache.eventmesh.common.protocol.tcp.UserAgent;
+import
org.apache.eventmesh.dashboard.core.function.SDK.config.CreateRuntimeConfig;
+
+import java.util.AbstractMap.SimpleEntry;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class RuntimeTcpEventMeshSDKOperationTest {
+
+ private final RuntimeTcpEventMeshSDKOperation eventMeshSDKOperation = new
RuntimeTcpEventMeshSDKOperation();
+
+ @Test
+ void testCreateClient() {
+ SimpleEntry<String, EventMeshMessageTCPClient> simpleEntry = null;
+ try {
+ final UserAgent userAgent = UserAgent.builder()
+ .env("test")
+ .host("localhost")
+ .password("123456")
+ .username("eventmesh")
+ .group("EventmeshTestGroup")
+ .path("/")
+ .port(8365)
+ .subsystem("501")
+ .pid(32893)
+ .version("2.1")
+ .idc("A")
+ .purpose(EventMeshCommon.USER_AGENT_PURPOSE_PUB)
+ .build();
+ log.info("userAgent {}", userAgent);
+ final CreateRuntimeConfig runtimeConfig =
CreateRuntimeConfig.builder()
+ .runtimeServerAddress("127.0.0.1:10000")
+ .userAgent(userAgent)
+ .build();
+ log.info("{}", runtimeConfig);
+ simpleEntry = eventMeshSDKOperation.createClient(runtimeConfig);
+ Assertions.assertEquals("127.0.0.1:10000", simpleEntry.getKey());
+ simpleEntry.getValue().close();
+ } catch (Exception e) {
+ log.error("create runtime tcp EventMeshMessage client failed", e);
+ if (simpleEntry != null) {
+ simpleEntry.getValue().close();
+ }
+ }
+ }
+}
diff --git
a/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/config/CreateRedisConfig.java
b/eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeTcpOpenMessageSDKOperationTest.java
similarity index 70%
copy from
eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/config/CreateRedisConfig.java
copy to
eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeTcpOpenMessageSDKOperationTest.java
index a457749..64300e0 100644
---
a/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/config/CreateRedisConfig.java
+++
b/eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeTcpOpenMessageSDKOperationTest.java
@@ -15,17 +15,18 @@
* limitations under the License.
*/
-package org.apache.eventmesh.dashboard.core.function.SDK.config;
+package org.apache.eventmesh.dashboard.core.function.SDK.operation;
-import lombok.Data;
-@Data
-public class CreateRedisConfig implements CreateSDKConfig {
+import org.junit.jupiter.api.Test;
- private String redisUrl;
+public class RuntimeTcpOpenMessageSDKOperationTest {
- @Override
- public String getUniqueKey() {
- return redisUrl;
+ private final RuntimeTcpOpenMessageSDKOperation tcpOpenMessageSDKOperation
= new RuntimeTcpOpenMessageSDKOperation();
+
+ @Test
+ void testCreateClient() {
+ // todo no impl
}
+
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
