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 b79395983 [ISSUE #3843]Simplify code in some subclasses of
AbstractTask and related class (#3844)
b79395983 is described below
commit b793959838f83eb8d7558c3f4ae8a4156c00dfaa
Author: pandaapo <[email protected]>
AuthorDate: Sat May 6 16:31:32 2023 +0800
[ISSUE #3843]Simplify code in some subclasses of AbstractTask and related
class (#3844)
* Some Optimization
* Fix checkstyle error.
* Simplify code.
* Revoke controversial import static.
---
.../core/protocol/tcp/client/session/Session.java | 128 +++++----------------
.../core/protocol/tcp/client/task/HelloTask.java | 3 +-
.../protocol/tcp/client/task/RecommendTask.java | 7 +-
.../protocol/tcp/client/task/UnSubscribeTask.java | 14 +--
4 files changed, 35 insertions(+), 117 deletions(-)
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 730d7a52b..0d07454e4 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
@@ -52,6 +52,8 @@ import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
+import lombok.Getter;
+import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
@Slf4j
@@ -61,112 +63,64 @@ public class Session {
private static final Logger SUBSCRIB_LOGGER =
LoggerFactory.getLogger("subscribeLogger");
+ @Setter
+ @Getter
private UserAgent client;
+ @Setter
+ @Getter
private InetSocketAddress remoteAddress;
+ @Setter
+ @Getter
protected ChannelHandlerContext context;
+ @Setter
+ @Getter
private WeakReference<ClientGroupWrapper> clientGroupWrapper;
+ @Setter
+ @Getter
private EventMeshTCPConfiguration eventMeshTCPConfiguration;
+ @Setter
+ @Getter
private SessionPusher pusher;
+ @Setter
+ @Getter
private SessionSender sender;
private final long createTime = System.currentTimeMillis();
+ @Setter
+ @Getter
private long lastHeartbeatTime = System.currentTimeMillis();
- private long isolateTime = 0;
+ @Setter
+ @Getter
+ private long isolateTime;
+ @Setter
+ @Getter
private SessionContext sessionContext = new SessionContext(this);
- private boolean listenRspSend = false;
+ private boolean listenRspSend;
private final ReentrantLock listenRspLock = new ReentrantLock();
- private String listenRequestSeq = null;
+ @Setter
+ @Getter
+ private String listenRequestSeq;
+ @Setter
+ @Getter
protected SessionState sessionState = SessionState.CREATED;
- public InetSocketAddress getRemoteAddress() {
- return remoteAddress;
- }
-
- public void setRemoteAddress(InetSocketAddress remoteAddress) {
- this.remoteAddress = remoteAddress;
- }
-
- public long getLastHeartbeatTime() {
- return lastHeartbeatTime;
- }
-
public void notifyHeartbeat(long heartbeatTime) throws Exception {
this.lastHeartbeatTime = heartbeatTime;
}
- public SessionState getSessionState() {
- return sessionState;
- }
-
- public void setSessionState(SessionState sessionState) {
- this.sessionState = sessionState;
- }
-
- public void setClient(UserAgent client) {
- this.client = client;
- }
-
- public SessionPusher getPusher() {
- return pusher;
- }
-
- public void setPusher(SessionPusher pusher) {
- this.pusher = pusher;
- }
-
- public SessionSender getSender() {
- return sender;
- }
-
- public void setSender(SessionSender sender) {
- this.sender = sender;
- }
-
- public void setLastHeartbeatTime(long lastHeartbeatTime) {
- this.lastHeartbeatTime = lastHeartbeatTime;
- }
-
- public SessionContext getSessionContext() {
- return sessionContext;
- }
-
- public void setSessionContext(SessionContext sessionContext) {
- this.sessionContext = sessionContext;
- }
-
- public ChannelHandlerContext getContext() {
- return context;
- }
-
- public void setContext(ChannelHandlerContext context) {
- this.context = context;
- }
-
- public UserAgent getClient() {
- return client;
- }
-
- public String getListenRequestSeq() {
- return listenRequestSeq;
- }
-
- public void setListenRequestSeq(String listenRequestSeq) {
- this.listenRequestSeq = listenRequestSeq;
- }
-
public void subscribe(List<SubscriptionItem> items) throws Exception {
for (SubscriptionItem item : items) {
sessionContext.getSubscribeTopics().putIfAbsent(item.getTopic(),
item);
@@ -299,14 +253,6 @@ public class Session {
return result;
}
- public WeakReference<ClientGroupWrapper> getClientGroupWrapper() {
- return clientGroupWrapper;
- }
-
- public void setClientGroupWrapper(WeakReference<ClientGroupWrapper>
clientGroupWrapper) {
- this.clientGroupWrapper = clientGroupWrapper;
- }
-
public Session(UserAgent client, ChannelHandlerContext context,
EventMeshTCPConfiguration eventMeshTCPConfiguration) {
this.client = client;
this.context = context;
@@ -316,14 +262,6 @@ public class Session {
this.pusher = new SessionPusher(this);
}
- public EventMeshTCPConfiguration getEventMeshTCPConfiguration() {
- return eventMeshTCPConfiguration;
- }
-
- public void setEventMeshTCPConfiguration(EventMeshTCPConfiguration
eventMeshTCPConfiguration) {
- this.eventMeshTCPConfiguration = eventMeshTCPConfiguration;
- }
-
public void trySendListenResponse(Header header, long startTime, long
taskExecuteTime) {
if (!listenRspSend && listenRspLock.tryLock()) {
if (header == null) {
@@ -340,14 +278,6 @@ public class Session {
}
}
- public long getIsolateTime() {
- return isolateTime;
- }
-
- public void setIsolateTime(long isolateTime) {
- this.isolateTime = isolateTime;
- }
-
public boolean isAvailable(String topic) {
if (SessionState.CLOSED == sessionState) {
log.warn("session is not available because session has been
closed,topic:{},client:{}", topic, client);
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/task/HelloTask.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/task/HelloTask.java
index 8e70bbd71..e024143cc 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/task/HelloTask.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/task/HelloTask.java
@@ -121,9 +121,8 @@ public class HelloTask extends AbstractTask {
throw new Exception("client version cannot be null");
}
- if (!(StringUtils.equals(EventMeshConstants.PURPOSE_PUB,
user.getPurpose()) || StringUtils.equals(
- EventMeshConstants.PURPOSE_SUB, user.getPurpose()))) {
+ if (!StringUtils.equalsAny(user.getPurpose(),
EventMeshConstants.PURPOSE_PUB, EventMeshConstants.PURPOSE_SUB)) {
throw new Exception("client purpose config is error");
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/task/RecommendTask.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/task/RecommendTask.java
index 078d218c6..b9cb7399f 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/task/RecommendTask.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/task/RecommendTask.java
@@ -36,10 +36,6 @@ import org.slf4j.LoggerFactory;
import io.netty.channel.ChannelHandlerContext;
-
-import lombok.extern.slf4j.Slf4j;
-
-@Slf4j
public class RecommendTask extends AbstractTask {
private static final Logger MESSAGE_LOGGER =
LoggerFactory.getLogger(EventMeshConstants.MESSAGE);
@@ -92,8 +88,7 @@ public class RecommendTask extends AbstractTask {
throw new Exception("client wemqPasswd cannot be null");
}
- if (!(StringUtils.equals(EventMeshConstants.PURPOSE_PUB,
user.getPurpose()) || StringUtils.equals(
- EventMeshConstants.PURPOSE_SUB, user.getPurpose()))) {
+ if (!StringUtils.equalsAny(user.getPurpose(),
EventMeshConstants.PURPOSE_PUB, EventMeshConstants.PURPOSE_SUB)) {
throw new Exception("client purpose config is error");
}
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/task/UnSubscribeTask.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/task/UnSubscribeTask.java
index ece2d15ce..b9793e032 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/task/UnSubscribeTask.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/task/UnSubscribeTask.java
@@ -30,17 +30,13 @@ import org.apache.commons.collections4.MapUtils;
import java.util.ArrayList;
import java.util.List;
-import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import io.netty.channel.ChannelHandlerContext;
-
-import lombok.extern.slf4j.Slf4j;
-
-@Slf4j
public class UnSubscribeTask extends AbstractTask {
private static final Logger MESSAGE_LOGGER =
LoggerFactory.getLogger(EventMeshConstants.MESSAGE);
@@ -55,11 +51,9 @@ public class UnSubscribeTask extends AbstractTask {
Package msg = new Package();
try {
synchronized (session) {
- List<SubscriptionItem> topics = new
ArrayList<SubscriptionItem>();
- if
(MapUtils.isNotEmpty(session.getSessionContext().getSubscribeTopics())) {
- for (Map.Entry<String, SubscriptionItem> entry :
session.getSessionContext().getSubscribeTopics().entrySet()) {
- topics.add(entry.getValue());
- }
+ ConcurrentHashMap<String, SubscriptionItem> subscribeTopics =
session.getSessionContext().getSubscribeTopics();
+ if (MapUtils.isNotEmpty(subscribeTopics)) {
+ List<SubscriptionItem> topics = new
ArrayList<>(subscribeTopics.values());
session.unsubscribe(topics);
MESSAGE_LOGGER.info("UnSubscriberTask
succeed|user={}|topics={}", session.getClient(), topics);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
