You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@eventmesh.apache.org by ch...@apache.org on 2022/08/29 03:10:57 UTC
[incubator-eventmesh] branch master updated: [Issue-1152] Extraction of constants [ClientGroupWrapper EventMeshProducer]
This is an automated email from the ASF dual-hosted git repository.
chenguangsheng pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-eventmesh.git
The following commit(s) were added to refs/heads/master by this push:
new 2533b876 [Issue-1152] Extraction of constants [ClientGroupWrapper EventMeshProducer]
new 1b94e38f Merge pull request #1190 from indhuvisu/issue-1152
2533b876 is described below
commit 2533b876922848fa9ebe33cb0994d6252154a4b6
Author: Indhumathi Viswanathan <in...@gmail.com>
AuthorDate: Sun Aug 28 20:20:39 2022 +0530
[Issue-1152] Extraction of constants [ClientGroupWrapper EventMeshProducer]
---
.../runtime/constants/EventMeshConstants.java | 11 +++--
.../protocol/grpc/producer/EventMeshProducer.java | 7 ++--
.../tcp/client/group/ClientGroupWrapper.java | 47 +++++++++++-----------
3 files changed, 36 insertions(+), 29 deletions(-)
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/constants/EventMeshConstants.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/constants/EventMeshConstants.java
index ee84c299..dd9ea0fc 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/constants/EventMeshConstants.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/constants/EventMeshConstants.java
@@ -76,8 +76,9 @@ public class EventMeshConstants {
public static final int DEFAULT_PUSH_RETRY_TIME_DISTANCE_IN_MILLSECONDS = 3000;
public static final String PURPOSE_PUB = "pub";
-
+ public static final String PURPOSE_PUB_UPPER_CASE = "PUB";
public static final String PURPOSE_SUB = "sub";
+ public static final String PURPOSE_SUB_UPPER_CASE = "SUB";
public static final String PURPOSE_ALL = "all";
@@ -105,7 +106,7 @@ public class EventMeshConstants {
public static final String MANAGE_GROUP = "group";
public static final String MANAGE_PURPOSE = "purpose";
public static final String MANAGE_TOPIC = "topic";
-
+ public static final String MANAGE_MSG = "msg";
public static final String EVENTMESH_SEND_BACK_TIMES = "eventmeshdendbacktimes";
public static final String EVENTMESH_SEND_BACK_IP = "eventmeshsendbackip";
@@ -129,6 +130,10 @@ public class EventMeshConstants {
public static final String LEAVE_TIME = "leave" + Constants.MESSAGE_PROP_SEPARATOR + "time"; //leaveBrokerTime
public static final String ARRIVE_TIME = "arrive" + Constants.MESSAGE_PROP_SEPARATOR + "time";
public static final String STORE_TIME = "store" + Constants.MESSAGE_PROP_SEPARATOR + "time";
-
+ public static final String PRODUCER_GROUP = "producerGroup";
+ public static final String CONSUMER_GROUP = "consumerGroup";
+ public static final String INSTANCE_NAME = "instanceName";
+ public static final String EVENT_MESH_IDC = "eventMeshIDC";
+ public static final String IS_BROADCAST = "isBroadcast";
}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/producer/EventMeshProducer.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/producer/EventMeshProducer.java
index 345576c8..1c2897f2 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/producer/EventMeshProducer.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/producer/EventMeshProducer.java
@@ -21,6 +21,7 @@ import org.apache.eventmesh.api.RequestReplyCallback;
import org.apache.eventmesh.api.SendCallback;
import org.apache.eventmesh.runtime.common.ServiceState;
import org.apache.eventmesh.runtime.configuration.EventMeshGrpcConfiguration;
+import org.apache.eventmesh.runtime.constants.EventMeshConstants;
import org.apache.eventmesh.runtime.core.consumergroup.ProducerGroupConf;
import org.apache.eventmesh.runtime.core.plugin.MQProducerWrapper;
import org.apache.eventmesh.runtime.util.EventMeshUtil;
@@ -59,12 +60,12 @@ public class EventMeshProducer {
this.producerGroupConfig = producerGroupConfig;
Properties keyValue = new Properties();
- keyValue.put("producerGroup", producerGroupConfig.getGroupName());
- keyValue.put("instanceName", EventMeshUtil.buildMeshClientID(
+ keyValue.put(EventMeshConstants.PRODUCER_GROUP, producerGroupConfig.getGroupName());
+ keyValue.put(EventMeshConstants.INSTANCE_NAME, EventMeshUtil.buildMeshClientID(
producerGroupConfig.getGroupName(), eventMeshGrpcConfiguration.eventMeshCluster));
//TODO for defibus
- keyValue.put("eventMeshIDC", eventMeshGrpcConfiguration.eventMeshIDC);
+ keyValue.put(EventMeshConstants.EVENT_MESH_IDC, eventMeshGrpcConfiguration.eventMeshIDC);
mqProducerWrapper = new MQProducerWrapper(
eventMeshGrpcConfiguration.eventMeshConnectorPluginType);
mqProducerWrapper.init(keyValue);
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientGroupWrapper.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientGroupWrapper.java
index fc85da4d..65317c3d 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientGroupWrapper.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientGroupWrapper.java
@@ -47,6 +47,7 @@ import org.apache.eventmesh.trace.api.common.EventMeshTraceConstants;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
+import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.Iterator;
@@ -276,12 +277,12 @@ public class ClientGroupWrapper {
}
Properties keyValue = new Properties();
- keyValue.put("producerGroup", group);
- keyValue.put("instanceName", EventMeshUtil
- .buildMeshTcpClientID(sysId, "PUB", eventMeshTCPConfiguration.eventMeshCluster));
+ keyValue.put(EventMeshConstants.PRODUCER_GROUP, group);
+ keyValue.put(EventMeshConstants.INSTANCE_NAME, EventMeshUtil
+ .buildMeshTcpClientID(sysId, EventMeshConstants.PURPOSE_PUB_UPPER_CASE, eventMeshTCPConfiguration.eventMeshCluster));
//TODO for defibus
- keyValue.put("eventMeshIDC", eventMeshTCPConfiguration.eventMeshIDC);
+ keyValue.put(EventMeshConstants.EVENT_MESH_IDC, eventMeshTCPConfiguration.eventMeshIDC);
mqProducerWrapper.init(keyValue);
mqProducerWrapper.start();
@@ -413,11 +414,11 @@ public class ClientGroupWrapper {
}
Properties keyValue = new Properties();
- keyValue.put("isBroadcast", "false");
- keyValue.put("consumerGroup", group);
- keyValue.put("eventMeshIDC", eventMeshTCPConfiguration.eventMeshIDC);
- keyValue.put("instanceName", EventMeshUtil
- .buildMeshTcpClientID(sysId, "SUB", eventMeshTCPConfiguration.eventMeshCluster));
+ keyValue.put(EventMeshConstants.IS_BROADCAST, "false");
+ keyValue.put(EventMeshConstants.CONSUMER_GROUP, group);
+ keyValue.put(EventMeshConstants.EVENT_MESH_IDC, eventMeshTCPConfiguration.eventMeshIDC);
+ keyValue.put(EventMeshConstants.INSTANCE_NAME, EventMeshUtil
+ .buildMeshTcpClientID(sysId, EventMeshConstants.PURPOSE_SUB_UPPER_CASE, eventMeshTCPConfiguration.eventMeshCluster));
persistentMsgConsumer.init(keyValue);
@@ -524,11 +525,11 @@ public class ClientGroupWrapper {
}
Properties keyValue = new Properties();
- keyValue.put("isBroadcast", "true");
- keyValue.put("consumerGroup", group);
- keyValue.put("eventMeshIDC", eventMeshTCPConfiguration.eventMeshIDC);
- keyValue.put("instanceName", EventMeshUtil
- .buildMeshTcpClientID(sysId, "SUB", eventMeshTCPConfiguration.eventMeshCluster));
+ keyValue.put(EventMeshConstants.IS_BROADCAST, "true");
+ keyValue.put(EventMeshConstants.CONSUMER_GROUP, group);
+ keyValue.put(EventMeshConstants.EVENT_MESH_IDC, eventMeshTCPConfiguration.eventMeshIDC);
+ keyValue.put(EventMeshConstants.INSTANCE_NAME, EventMeshUtil
+ .buildMeshTcpClientID(sysId, EventMeshConstants.PURPOSE_SUB_UPPER_CASE, eventMeshTCPConfiguration.eventMeshCluster));
broadCastMsgConsumer.init(keyValue);
EventListener listener = (event, context) -> {
@@ -699,19 +700,19 @@ public class ClientGroupWrapper {
try {
logger.info("pushMsgToEventMesh,targetUrl:{},msg:{}", targetUrl,
- msg);
+ msg);
List<String> paramValues = new ArrayList<String>();
- paramValues.add("msg");
+ paramValues.add(EventMeshConstants.MANAGE_MSG);
paramValues.add(JsonUtils.serialize(msg));
- paramValues.add("group");
+ paramValues.add(EventMeshConstants.MANAGE_GROUP);
paramValues.add(group);
result = HttpTinyClient.httpPost(
- targetUrl.toString(),
- null,
- paramValues,
- "UTF-8",
- 3000);
+ targetUrl.toString(),
+ null,
+ paramValues,
+ StandardCharsets.UTF_8.name(),
+ 3000);
} catch (Exception e) {
logger.error("httpPost " + targetUrl + " is fail,", e);
throw e;
@@ -753,7 +754,7 @@ public class ClientGroupWrapper {
+ " topic:{}", group, bizSeqNo, topic);
}
- });
+ });
eventMeshTcpMonitor.getTcpSummaryMetrics().getEventMesh2mqMsgNum().incrementAndGet();
} catch (Exception e) {
logger.warn("try send msg back to broker failed");
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: commits-help@eventmesh.apache.org