[ISSUE #3348]Modify SessionContext's public attributes to private

mxsm · mxsm · commit af8a6cdf9d9a · 2023-03-06T00:01:48.000+08:00
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
@@ -277,7 +277,7 @@ private void cleanClientGroupWrapperByClosePub(Session session) throws Exception
      * @param session
      */
     private void cleanSubscriptionInSession(Session session) throws Exception {
-        for (SubscriptionItem item : session.getSessionContext().subscribeTopics.values()) {
+        for (SubscriptionItem item : session.getSessionContext().getSubscribeTopics().values()) {
             ClientGroupWrapper clientGroupWrapper = Objects.requireNonNull(session.getClientGroupWrapper().get());
             clientGroupWrapper.removeSubscription(item, session);
             if (!clientGroupWrapper.hasSubscription(item.getTopic())) {
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
@@ -169,7 +169,7 @@ public void setListenRequestSeq(String listenRequestSeq) {
 
     public void subscribe(List<SubscriptionItem> items) throws Exception {
         for (SubscriptionItem item : items) {
-            sessionContext.subscribeTopics.putIfAbsent(item.getTopic(), item);
+            sessionContext.getSubscribeTopics().putIfAbsent(item.getTopic(), item);
             Objects.requireNonNull(clientGroupWrapper.get()).subscribe(item);
 
             Objects.requireNonNull(clientGroupWrapper.get()).getMqProducerWrapper().getMeshMQProducer()
@@ -182,7 +182,7 @@ public void subscribe(List<SubscriptionItem> items) throws Exception {
 
     public void unsubscribe(List<SubscriptionItem> items) throws Exception {
         for (SubscriptionItem item : items) {
-            sessionContext.subscribeTopics.remove(item.getTopic());
+            sessionContext.getSubscribeTopics().remove(item.getTopic());
             Objects.requireNonNull(clientGroupWrapper.get()).removeSubscription(item, this);
 
             if (!Objects.requireNonNull(clientGroupWrapper.get()).hasSubscription(item.getTopic())) {
@@ -195,7 +195,7 @@ public void unsubscribe(List<SubscriptionItem> items) throws Exception {
     public EventMeshTcpSendResult upstreamMsg(Header header, CloudEvent event, SendCallback sendCallback,
         long startTime, long taskExecuteTime) {
         String topic = event.getSubject();
-        sessionContext.sendTopics.putIfAbsent(topic, topic);
+        sessionContext.getSendTopics().putIfAbsent(topic, topic);
         return sender.send(header, event, sendCallback, startTime, taskExecuteTime);
     }
 
@@ -358,7 +358,7 @@ public boolean isAvailable(String topic) {
             return false;
         }
 
-        if (!sessionContext.subscribeTopics.containsKey(topic)) {
+        if (!sessionContext.getSubscribeTopics().containsKey(topic)) {
             log.warn("session is not available because session has not subscribe topic:{},client:{}", topic, client);
             return false;
         }
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/session/SessionContext.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/session/SessionContext.java
@@ -24,13 +24,17 @@
 
 import java.util.concurrent.ConcurrentHashMap;
 
+import lombok.Getter;
+
 public class SessionContext {
 
     private Session session;
 
-    public ConcurrentHashMap<String, String> sendTopics = new ConcurrentHashMap<String, String>();
+    @Getter
+    private final ConcurrentHashMap<String, String> sendTopics = new ConcurrentHashMap<>(64);
 
-    public ConcurrentHashMap<String, SubscriptionItem> subscribeTopics = new ConcurrentHashMap<String, SubscriptionItem>();
+    @Getter
+    private final ConcurrentHashMap<String/*Topic*/, SubscriptionItem> subscribeTopics = new ConcurrentHashMap<>(64);
 
     public long createTime = System.currentTimeMillis();
 
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
@@ -55,8 +55,8 @@ public void run() {
         try {
             synchronized (session) {
                 List<SubscriptionItem> topics = new ArrayList<SubscriptionItem>();
-                if (MapUtils.isNotEmpty(session.getSessionContext().subscribeTopics)) {
-                    for (Map.Entry<String, SubscriptionItem> entry : session.getSessionContext().subscribeTopics.entrySet()) {
+                if (MapUtils.isNotEmpty(session.getSessionContext().getSubscribeTopics())) {
+                    for (Map.Entry<String, SubscriptionItem> entry : session.getSessionContext().getSubscribeTopics().entrySet()) {
                         topics.add(entry.getValue());
                     }
                     session.unsubscribe(topics);
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/tcp/EventMeshTcpMonitor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/tcp/EventMeshTcpMonitor.java
@@ -112,14 +112,14 @@ public void start() throws Exception {
                 AtomicLong deliveredMsgsCount = session.getPusher().getDeliveredMsgsCount();
                 AtomicLong deliveredFailCount = session.getPusher().getDeliverFailMsgsCount();
                 int unAckMsgsCount = session.getPusher().getTotalUnackMsgs();
-                int sendTopics = session.getSessionContext().sendTopics.size();
-                int subscribeTopics = session.getSessionContext().subscribeTopics.size();
+                int sendTopics = session.getSessionContext().getSendTopics().size();
+                int subscribeTopics = session.getSessionContext().getSubscribeTopics().size();
 
                 tcpLogger.info("session|deliveredFailCount={}|deliveredMsgsCount={}|unAckMsgsCount={}|sendTopics={}|subscribeTopics={}|user={}",
                     deliveredFailCount.longValue(), deliveredMsgsCount.longValue(),
                     unAckMsgsCount, sendTopics, subscribeTopics, session.getClient());
 
-                topicSet.addAll(session.getSessionContext().subscribeTopics.keySet());
+                topicSet.addAll(session.getSessionContext().getSubscribeTopics().keySet());
             }
             tcpSummaryMetrics.setSubTopicNum(topicSet.size());
             tcpSummaryMetrics.setAllConnections(EventMeshTcpConnectionHandler.connections.get());
