This is an automated email from the ASF dual-hosted git repository.
mikexue 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 1600c6bf1 [ISSUE #4144] Subscription is almost impossible to be
cancelled when the TCP sub client is closed (#4145)
1600c6bf1 is described below
commit 1600c6bf1cbdb34de4bc3e9630898ab158c7af6f
Author: pandaapo <[email protected]>
AuthorDate: Wed Jun 28 14:36:53 2023 +0800
[ISSUE #4144] Subscription is almost impossible to be cancelled when the
TCP sub client is closed (#4145)
* Fix bug: suscription session can't be removed correctly.
* Fix bug: suscription session can't be removed correctly.
* remove the modification for other issue.
---
.../handler/ShowListenClientByTopicHandler.java | 7 +++--
.../tcp/client/group/ClientGroupWrapper.java | 31 +++++++++++-----------
.../core/protocol/tcp/client/session/Session.java | 5 ++++
3 files changed, 23 insertions(+), 20 deletions(-)
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/ShowListenClientByTopicHandler.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/ShowListenClientByTopicHandler.java
index 10d600182..8d9376f46 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/ShowListenClientByTopicHandler.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/ShowListenClientByTopicHandler.java
@@ -31,7 +31,6 @@ import
org.apache.eventmesh.runtime.core.protocol.tcp.client.session.Session;
import java.io.IOException;
import java.io.OutputStream;
import java.util.Map;
-import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
@@ -67,10 +66,10 @@ public class ShowListenClientByTopicHandler extends
AbstractHttpHandler {
ConcurrentHashMap<String, ClientGroupWrapper> clientGroupMap =
clientSessionGroupMapping.getClientGroupMap();
if (!clientGroupMap.isEmpty()) {
for (ClientGroupWrapper cgw : clientGroupMap.values()) {
- Set<Session> listenSessionSet =
cgw.getTopic2sessionInGroupMapping().get(topic);
- if (listenSessionSet != null &&
!listenSessionSet.isEmpty()) {
+ Map<String, Session> listenSessions =
cgw.getTopic2sessionInGroupMapping().get(topic);
+ if (listenSessions != null && !listenSessions.isEmpty()) {
result.append(String.format("group:%s",
cgw.getGroup())).append(newLine);
- for (Session session : listenSessionSet) {
+ for (Session session : listenSessions.values()) {
UserAgent userAgent = session.getClient();
result.append(String.format("pid=%s | ip=%s |
port=%s | path=%s | version=%s", userAgent.getPid(), userAgent
.getHost(), userAgent.getPort(),
userAgent.getPath(), userAgent.getVersion()))
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 097f89144..72491a2ed 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
@@ -49,9 +49,11 @@ import org.apache.commons.lang3.StringUtils;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
+import java.util.Map;
import java.util.Objects;
import java.util.Properties;
import java.util.Set;
@@ -114,8 +116,8 @@ public class ClientGroupWrapper {
private MQConsumerWrapper broadCastMsgConsumer;
- private final ConcurrentHashMap<String, Set<Session>>
topic2sessionInGroupMapping =
- new ConcurrentHashMap<String, Set<Session>>();
+ private final ConcurrentHashMap<String, Map<String, Session>>
topic2sessionInGroupMapping =
+ new ConcurrentHashMap<String, Map<String, Session>>();
private final ConcurrentHashMap<String, SubscriptionItem> subscriptions =
new ConcurrentHashMap<>();
@@ -137,7 +139,7 @@ public class ClientGroupWrapper {
this.mqProducerWrapper = new
MQProducerWrapper(eventMeshTCPServer.getEventMeshTCPConfiguration().getEventMeshStoragePluginType());
}
- public ConcurrentHashMap<String, Set<Session>>
getTopic2sessionInGroupMapping() {
+ public ConcurrentHashMap<String, Map<String, Session>>
getTopic2sessionInGroupMapping() {
return topic2sessionInGroupMapping;
}
@@ -206,24 +208,22 @@ public class ClientGroupWrapper {
return false;
}
- boolean r;
+ boolean r = false;
try {
this.groupLock.writeLock().lockInterruptibly();
if (!topic2sessionInGroupMapping.containsKey(topic)) {
- Set<Session> sessions = new HashSet<Session>();
+ Map<String, Session> sessions = new HashMap<>();
topic2sessionInGroupMapping.put(topic, sessions);
}
- r = topic2sessionInGroupMapping.get(topic).add(session);
- if (r) {
-
+ if (r =
topic2sessionInGroupMapping.get(topic).putIfAbsent(session.getSessionId(),
session) == null) {
if (log.isInfoEnabled()) {
- log.info("addSubscription success, group:{} topic:{}
client:{}", group,
- topic, session.getClient());
+ log.info("Cache session success, group:{} topic:{}
client:{} sessionId:{}", group,
+ topic, session.getClient(), session.getSessionId());
}
} else {
if (log.isWarnEnabled()) {
- log.warn("addSubscription fail, group:{} topic:{}
client:{}", group, topic,
- session.getClient());
+ log.warn("Session already exists in
topic2sessionInGroupMapping. group:{} topic:{} client:{} sessionId:{}", group,
topic,
+ session.getClient(), session.getSessionId());
}
}
@@ -257,8 +257,7 @@ public class ClientGroupWrapper {
try {
this.groupLock.writeLock().lockInterruptibly();
if (topic2sessionInGroupMapping.containsKey(topic)) {
- r = topic2sessionInGroupMapping.get(topic).remove(session);
- if (r) {
+ if
(topic2sessionInGroupMapping.get(topic).remove(session.getSessionId()) != null)
{
if (log.isInfoEnabled()) {
log.info(
@@ -268,8 +267,8 @@ public class ClientGroupWrapper {
} else {
if (log.isWarnEnabled()) {
log.warn(
- "removeSubscription remove session failed,
group:{} topic:{} client:{}",
- group, topic, session.getClient());
+ "Not found session in cache, group:{} topic:{}
client:{} sessionId:{}",
+ group, topic, session.getClient(),
session.getSessionId());
}
}
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/session/Session.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/session/Session.java
index 0d07454e4..c5126a630 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/session/Session.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/session/Session.java
@@ -41,6 +41,7 @@ import java.lang.ref.WeakReference;
import java.net.InetSocketAddress;
import java.util.List;
import java.util.Objects;
+import java.util.UUID;
import java.util.concurrent.locks.ReentrantLock;
import org.slf4j.Logger;
@@ -117,6 +118,10 @@ public class Session {
@Getter
protected SessionState sessionState = SessionState.CREATED;
+ @Setter
+ @Getter
+ private String sessionId = UUID.randomUUID().toString();
+
public void notifyHeartbeat(long heartbeatTime) throws Exception {
this.lastHeartbeatTime = heartbeatTime;
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
