This is an automated email from the ASF dual-hosted git repository.
mxsm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/eventmesh.git
The following commit(s) were added to refs/heads/master by this push:
new 927bc5bcc [ISSUE #4192] Do some code optimization.[StreamTopicConfig]
(#4464)
927bc5bcc is described below
commit 927bc5bcc8746c118d2178fce717179544d5e2cb
Author: PAN <[email protected]>
AuthorDate: Tue Sep 26 13:42:08 2023 +0800
[ISSUE #4192] Do some code optimization.[StreamTopicConfig] (#4464)
---
.../grpc/consumer/consumergroup/StreamTopicConfig.java | 18 +++++++++++-------
1 file changed, 11 insertions(+), 7 deletions(-)
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/consumergroup/StreamTopicConfig.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/consumergroup/StreamTopicConfig.java
index 36632466f..65699d9f1 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/consumergroup/StreamTopicConfig.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/consumergroup/StreamTopicConfig.java
@@ -40,19 +40,23 @@ public class StreamTopicConfig extends
ConsumerGroupTopicConfig {
/**
* Key: IDC Value: list of emitters with Client_IP:port
*/
- private final transient Map<String, Map<String, EventEmitter<CloudEvent>>>
idcEmitterMap = new ConcurrentHashMap<>();
+ private final Map<String, Map<String, EventEmitter<CloudEvent>>>
idcEmitterMap = new ConcurrentHashMap<>();
/**
* Key: IDC Value: list of emitters
*/
- private transient Map<String, List<EventEmitter<CloudEvent>>> idcEmitters
= new ConcurrentHashMap<>();
+ private Map<String, List<EventEmitter<CloudEvent>>> idcEmitters = new
ConcurrentHashMap<>();
- private transient List<EventEmitter<CloudEvent>> totalEmitters = new
ArrayList<>();
+ private List<EventEmitter<CloudEvent>> totalEmitters = new ArrayList<>();
public StreamTopicConfig(final String consumerGroup, final String topic,
final SubscriptionMode subscriptionMode) {
super(consumerGroup, topic, subscriptionMode, GrpcType.STREAM);
}
+ private String concatKey(String ip, String pid) {
+ return ip.concat(":").concat(pid);
+ }
+
@Override
public synchronized void registerClient(final ConsumerGroupClient client) {
Objects.requireNonNull(client, "ConsumerGroupClient can not be null");
@@ -66,7 +70,7 @@ public class StreamTopicConfig extends
ConsumerGroupTopicConfig {
}
idcEmitterMap.computeIfAbsent(client.getIdc(), k -> new HashMap<>())
- .put(client.getIp() + ":" + client.getPid(),
client.getEventEmitter());
+ .put(concatKey(client.getIp(), client.getPid()),
client.getEventEmitter());
idcEmitters = buildIdcEmitter(idcEmitterMap);
totalEmitters = buildTotalEmitter(idcEmitters);
@@ -119,9 +123,7 @@ public class StreamTopicConfig extends
ConsumerGroupTopicConfig {
private static Map<String, List<EventEmitter<CloudEvent>>> buildIdcEmitter(
final Map<String, Map<String, EventEmitter<CloudEvent>>>
idcEmitterMap) {
final Map<String, List<EventEmitter<CloudEvent>>> result = new
HashMap<>();
- idcEmitterMap.forEach((k, v) -> {
- result.put(k, new
LinkedList<EventEmitter<CloudEvent>>(v.values()));
- });
+ idcEmitterMap.forEach((k, v) -> result.put(k, new
LinkedList<>(v.values())));
return result;
}
@@ -132,3 +134,5 @@ public class StreamTopicConfig extends
ConsumerGroupTopicConfig {
return emitterList;
}
}
+
+
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
