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 12aeb4bbb [ISSUE #4109 ] Code Optimization.[ClientGroupWrapper] (#4154)
12aeb4bbb is described below
commit 12aeb4bbbc6156d80057b33ef3a558a60b6ea516
Author: Nitheesh Daram <[email protected]>
AuthorDate: Tue Jun 27 15:20:46 2023 +0530
[ISSUE #4109 ] Code Optimization.[ClientGroupWrapper] (#4154)
* Resolves #4109
Signed-off-by: Nitheesh Daram
<[email protected]>
* Resolves #4109
Code Review Changes
Signed-off-by: Nitheesh Daram
<[email protected]>
---------
Signed-off-by: Nitheesh Daram
<[email protected]>
---
.../tcp/client/group/ClientGroupWrapper.java | 52 +++++++++++++++-------
.../client/group/ClientSessionGroupMapping.java | 20 ++++-----
2 files changed, 47 insertions(+), 25 deletions(-)
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 ac45ca6cb..097f89144 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
@@ -66,6 +66,7 @@ import io.opentelemetry.api.trace.Span;
import com.google.common.base.Preconditions;
+import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
@@ -88,17 +89,26 @@ public class ClientGroupWrapper {
private final ReadWriteLock groupLock = new ReentrantReadWriteLock();
- public Set<Session> groupConsumerSessions = new HashSet<Session>();
+ @Getter
+ private final Set<Session> groupConsumerSessions = new HashSet<Session>();
- public Set<Session> groupProducerSessions = new HashSet<Session>();
+ @Getter
+ private final Set<Session> groupProducerSessions = new HashSet<Session>();
- public AtomicBoolean started4Persistent = new AtomicBoolean(Boolean.FALSE);
+ @Getter
+ private final AtomicBoolean started4Persistent = new
AtomicBoolean(Boolean.FALSE);
- public AtomicBoolean started4Broadcast = new AtomicBoolean(Boolean.FALSE);
+ @Getter
+ private final AtomicBoolean started4Broadcast = new
AtomicBoolean(Boolean.FALSE);
- public AtomicBoolean inited4Persistent = new AtomicBoolean(Boolean.FALSE);
+ @Getter
+ private final AtomicBoolean inited4Persistent = new
AtomicBoolean(Boolean.FALSE);
- public AtomicBoolean inited4Broadcast = new AtomicBoolean(Boolean.FALSE);
+ @Getter
+ private final AtomicBoolean inited4Broadcast = new
AtomicBoolean(Boolean.FALSE);
+
+ @Getter
+ private final AtomicBoolean producerStarted = new
AtomicBoolean(Boolean.FALSE);
private MQConsumerWrapper persistentMsgConsumer;
@@ -109,8 +119,6 @@ public class ClientGroupWrapper {
private final ConcurrentHashMap<String, SubscriptionItem> subscriptions =
new ConcurrentHashMap<>();
- public AtomicBoolean producerStarted = new AtomicBoolean(Boolean.FALSE);
-
private final MQProducerWrapper mqProducerWrapper;
public ClientGroupWrapper(String sysId, String group,
@@ -139,6 +147,9 @@ public class ClientGroupWrapper {
this.groupLock.readLock().lockInterruptibly();
has = topic2sessionInGroupMapping.containsKey(topic);
} catch (Exception e) {
+ if (e instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
log.error("hasSubscription error! topic[{}]", topic);
} finally {
this.groupLock.readLock().unlock();
@@ -218,6 +229,9 @@ public class ClientGroupWrapper {
subscriptions.putIfAbsent(topic, subscriptionItem);
} catch (Exception e) {
+ if (e instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
log.error("addSubscription error! topic:{} client:{}", topic,
session.getClient(), e);
throw new Exception("addSubscription fail");
} finally {
@@ -267,6 +281,9 @@ public class ClientGroupWrapper {
group, topic);
}
} catch (Exception e) {
+ if (e instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
log.error("removeSubscription error! topic:{} client:{}", topic,
session.getClient(),
e);
} finally {
@@ -333,6 +350,9 @@ public class ClientGroupWrapper {
}
}
} catch (Exception e) {
+ if (e instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
log.error("addGroupConsumerSession error! group:{} client:{}",
group,
session.getClient(), e);
} finally {
@@ -360,6 +380,9 @@ public class ClientGroupWrapper {
session.getClient());
}
} catch (Exception e) {
+ if (e instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
log.error("addGroupProducerSession error! group:{} client:{}",
group,
session.getClient(), e);
} finally {
@@ -389,6 +412,9 @@ public class ClientGroupWrapper {
}
}
} catch (Exception e) {
+ if (e instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
log.error("removeGroupConsumerSession error! group:{} client:{}",
group,
session.getClient(), e);
} finally {
@@ -415,6 +441,9 @@ public class ClientGroupWrapper {
session.getClient());
}
} catch (Exception e) {
+ if (e instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
log.error("removeGroupProducerSession error! group:{} client:{}",
group,
session.getClient(), e);
} finally {
@@ -670,13 +699,6 @@ public class ClientGroupWrapper {
persistentMsgConsumer = null;
}
- public Set<Session> getGroupConsumerSessions() {
- return groupConsumerSessions;
- }
-
- public Set<Session> getGroupProducerSessions() {
- return groupProducerSessions;
- }
public EventMeshTCPConfiguration getEventMeshTCPConfiguration() {
return eventMeshTCPConfiguration;
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientSessionGroupMapping.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientSessionGroupMapping.java
index a0c088008..f8873ca9a 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientSessionGroupMapping.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientSessionGroupMapping.java
@@ -209,7 +209,7 @@ public class ClientSessionGroupMapping {
}
private void startClientGroupProducer(ClientGroupWrapper cgw, Session
session) throws Exception {
- if (!cgw.producerStarted.get()) {
+ if (!cgw.getProducerStarted().get()) {
cgw.startClientGroupProducer();
}
boolean flag = cgw.addGroupProducerSession(session);
@@ -220,15 +220,15 @@ public class ClientSessionGroupMapping {
}
private void initClientGroupConsumser(ClientGroupWrapper cgw) throws
Exception {
- if (!cgw.producerStarted.get()) {
+ if (!cgw.getProducerStarted().get()) {
cgw.startClientGroupProducer();
}
- if (!cgw.inited4Broadcast.get()) {
+ if (!cgw.getInited4Broadcast().get()) {
cgw.initClientGroupBroadcastConsumer();
}
- if (!cgw.inited4Persistent.get()) {
+ if (!cgw.getInited4Persistent().get()) {
cgw.initClientGroupPersistentConsumer();
}
}
@@ -247,10 +247,10 @@ public class ClientSessionGroupMapping {
throw new Exception("addGroupConsumerSession fail");
}
- if (cgw.inited4Persistent.get() && !cgw.started4Persistent.get()) {
+ if (cgw.getInited4Persistent().get() &&
!cgw.getStarted4Persistent().get()) {
cgw.startClientGroupPersistentConsumer();
}
- if (cgw.inited4Broadcast.get() && !cgw.started4Broadcast.get()) {
+ if (cgw.getInited4Broadcast().get() &&
!cgw.getStarted4Broadcast().get()) {
cgw.startClientGroupBroadcastConsumer();
}
session.setSessionState(SessionState.RUNNING);
@@ -306,7 +306,7 @@ public class ClientSessionGroupMapping {
Session reChooseSession =
clientGroupWrapper.getDownstreamDispatchStrategy()
.select(clientGroupWrapper.getGroup(),
downStreamMsgContext.event.getSubject(),
- clientGroupWrapper.groupConsumerSessions);
+ clientGroupWrapper.getGroupConsumerSessions());
if (reChooseSession != null) {
downStreamMsgContext.setSession(reChooseSession);
reChooseSession.getPusher().unAckMsg(downStreamMsgContext.seq,
downStreamMsgContext);
@@ -340,18 +340,18 @@ public class ClientSessionGroupMapping {
}
private void shutdownClientGroupConsumer(ClientGroupWrapper
clientGroupWrapper) throws Exception {
- if (clientGroupWrapper.started4Broadcast.get()) {
+ if (clientGroupWrapper.getStarted4Broadcast().get()) {
clientGroupWrapper.shutdownBroadCastConsumer();
}
- if (clientGroupWrapper.started4Persistent.get()) {
+ if (clientGroupWrapper.getStarted4Persistent().get()) {
clientGroupWrapper.shutdownPersistentConsumer();
}
}
private void shutdownClientGroupProducer(ClientGroupWrapper
clientGroupWrapper) throws Exception {
- if (clientGroupWrapper.producerStarted.get()) {
+ if (clientGroupWrapper.getProducerStarted().get()) {
clientGroupWrapper.shutdownProducer();
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
