[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
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.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 @@
*/
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 @@
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 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 @@
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.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 @@
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 @@
}
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.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 @@
@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.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/RuntimeGrpcConsumerSDKOperation.java b/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeGrpcConsumerSDKOperation.java
new file mode 100644
index 0000000..63e26a3
--- /dev/null
+++ b/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeGrpcConsumerSDKOperation.java
@@ -0,0 +1,54 @@
+/*
+ * 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.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.CreateRuntimeConfig;
+import org.apache.eventmesh.dashboard.core.function.SDK.config.CreateSDKConfig;
+
+import java.util.AbstractMap.SimpleEntry;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class RuntimeGrpcConsumerSDKOperation extends AbstractSDKOperation<EventMeshGrpcConsumer> {
+
+ @Override
+ public SimpleEntry<String, EventMeshGrpcConsumer> createClient(CreateSDKConfig clientConfig) {
+ final CreateRuntimeConfig runtimeConfig = (CreateRuntimeConfig) clientConfig;
+ final EventMeshGrpcClientConfig grpcClientConfig = buildEventMeshGrpcConsumerConfig(runtimeConfig);
+ EventMeshGrpcConsumer grpcConsumer = null;
+ try {
+ grpcConsumer = new EventMeshGrpcConsumer(grpcClientConfig);
+ grpcConsumer.init();
+ } catch (EventMeshException e) {
+ log.error("create runtime grpc Consumer client failed", e);
+ }
+ return new SimpleEntry<>(clientConfig.getUniqueKey(), grpcConsumer);
+ }
+
+ @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/RuntimeGrpcProducerSDKOperation.java b/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeGrpcProducerSDKOperation.java
new file mode 100644
index 0000000..2c46a6b
--- /dev/null
+++ b/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeGrpcProducerSDKOperation.java
@@ -0,0 +1,53 @@
+/*
+ * 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.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.CreateRuntimeConfig;
+import org.apache.eventmesh.dashboard.core.function.SDK.config.CreateSDKConfig;
+
+import java.util.AbstractMap.SimpleEntry;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class RuntimeGrpcProducerSDKOperation extends AbstractSDKOperation<EventMeshGrpcProducer> {
+
+ @Override
+ public SimpleEntry<String, EventMeshGrpcProducer> createClient(CreateSDKConfig clientConfig) {
+ final CreateRuntimeConfig runtimeConfig = (CreateRuntimeConfig) clientConfig;
+ final EventMeshGrpcClientConfig grpcClientConfig = buildEventMeshGrpcProducerConfig(runtimeConfig);
+ EventMeshGrpcProducer grpcProducer = null;
+ try {
+ grpcProducer = new EventMeshGrpcProducer(grpcClientConfig);
+ } catch (EventMeshException e) {
+ log.error("create runtime grpc Producer client failed", e);
+ }
+ return new SimpleEntry<>(clientConfig.getUniqueKey(), grpcProducer);
+ }
+
+ @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/RuntimeHttpConsumerSDKOperation.java b/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeHttpConsumerSDKOperation.java
new file mode 100644
index 0000000..e6c026a
--- /dev/null
+++ b/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeHttpConsumerSDKOperation.java
@@ -0,0 +1,53 @@
+/*
+ * 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.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.CreateRuntimeConfig;
+import org.apache.eventmesh.dashboard.core.function.SDK.config.CreateSDKConfig;
+
+import java.util.AbstractMap.SimpleEntry;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class RuntimeHttpConsumerSDKOperation extends AbstractSDKOperation<EventMeshHttpConsumer> {
+
+ @Override
+ public SimpleEntry<String, EventMeshHttpConsumer> createClient(CreateSDKConfig clientConfig) {
+ final CreateRuntimeConfig runtimeConfig = (CreateRuntimeConfig) clientConfig;
+ final EventMeshHttpClientConfig httpClientConfig = buildEventMeshHttpConsumerConfig(runtimeConfig);
+ EventMeshHttpConsumer httpConsumer = null;
+ try {
+ httpConsumer = new EventMeshHttpConsumer(httpClientConfig);
+ } catch (EventMeshException e) {
+ log.error("create runtime http Consumer client failed", e);
+ }
+ return new SimpleEntry<>(clientConfig.getUniqueKey(), httpConsumer);
+ }
+
+ @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/RuntimeHttpProducerSDKOperation.java b/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeHttpProducerSDKOperation.java
new file mode 100644
index 0000000..79071c7
--- /dev/null
+++ b/eventmesh-dashboard-core/src/main/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeHttpProducerSDKOperation.java
@@ -0,0 +1,53 @@
+/*
+ * 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.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.CreateRuntimeConfig;
+import org.apache.eventmesh.dashboard.core.function.SDK.config.CreateSDKConfig;
+
+import java.util.AbstractMap.SimpleEntry;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class RuntimeHttpProducerSDKOperation extends AbstractSDKOperation<EventMeshHttpProducer> {
+
+ @Override
+ public SimpleEntry<String, EventMeshHttpProducer> createClient(CreateSDKConfig clientConfig) {
+ final CreateRuntimeConfig runtimeConfig = (CreateRuntimeConfig) clientConfig;
+ final EventMeshHttpClientConfig httpClientConfig = buildEventMeshHttpProducerConfig(runtimeConfig);
+ EventMeshHttpProducer httpProducer = null;
+ try {
+ httpProducer = new EventMeshHttpProducer(httpClientConfig);
+ } catch (EventMeshException e) {
+ log.error("create runtime http Producer client failed", e);
+ }
+ return new SimpleEntry<>(clientConfig.getUniqueKey(), httpProducer);
+ }
+
+ @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/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.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 @@
@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 @@
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 @@
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 @@
@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/RuntimeGrpcConsumerSDKOperationTest.java b/eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeGrpcConsumerSDKOperationTest.java
new file mode 100644
index 0000000..a3869a6
--- /dev/null
+++ b/eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeGrpcConsumerSDKOperationTest.java
@@ -0,0 +1,56 @@
+/*
+ * 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.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 lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class RuntimeGrpcConsumerSDKOperationTest {
+
+ private final RuntimeGrpcConsumerSDKOperation grpcConsumerSDKOperation = new RuntimeGrpcConsumerSDKOperation();
+
+ @Test
+ void testCreateClient() {
+ SimpleEntry<String, EventMeshGrpcConsumer> grpcConsumerSimpleEntry = null;
+ try {
+ 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 runtime GRPC consumer client failed", e);
+ if (grpcConsumerSimpleEntry != null) {
+ grpcConsumerSimpleEntry.getValue().close();
+ }
+ }
+ }
+}
diff --git a/eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeGrpcProducerSDKOperationTest.java b/eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeGrpcProducerSDKOperationTest.java
new file mode 100644
index 0000000..f593f2f
--- /dev/null
+++ b/eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeGrpcProducerSDKOperationTest.java
@@ -0,0 +1,56 @@
+/*
+ * 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.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 lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class RuntimeGrpcProducerSDKOperationTest {
+
+ private final RuntimeGrpcProducerSDKOperation grpcProducerSDKOperation = new RuntimeGrpcProducerSDKOperation();
+
+ @Test
+ void testCreateClient() {
+ SimpleEntry<String, EventMeshGrpcProducer> grpcProducerSimpleEntry = null;
+ try {
+ 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 runtime GRPC producer client failed", e);
+ if (grpcProducerSimpleEntry != null) {
+ grpcProducerSimpleEntry.getValue().close();
+ }
+ }
+ }
+}
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/test/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeTcpOpenMessageSDKOperationTest.java b/eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeTcpOpenMessageSDKOperationTest.java
new file mode 100644
index 0000000..64300e0
--- /dev/null
+++ b/eventmesh-dashboard-core/src/test/java/org/apache/eventmesh/dashboard/core/function/SDK/operation/RuntimeTcpOpenMessageSDKOperationTest.java
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.dashboard.core.function.SDK.operation;
+
+
+import org.junit.jupiter.api.Test;
+
+public class RuntimeTcpOpenMessageSDKOperationTest {
+
+ private final RuntimeTcpOpenMessageSDKOperation tcpOpenMessageSDKOperation = new RuntimeTcpOpenMessageSDKOperation();
+
+ @Test
+ void testCreateClient() {
+ // todo no impl
+ }
+
+}
