[waterflow] fix: memory leak in ConditionFrom

Author: loveTsong

framework/waterflow/java/waterflow-core/src/main/java/modelengine/fit/waterflow/domain/stream/nodes/ConditionsNode.java

@@ -11,15 +11,9 @@
 import modelengine.fit.waterflow.domain.context.repo.flowcontext.FlowContextRepo;
 import modelengine.fit.waterflow.domain.context.repo.flowlock.FlowLocks;
 import modelengine.fit.waterflow.domain.enums.FlowNodeType;
-import modelengine.fit.waterflow.domain.stream.reactive.Subscription;
-import modelengine.fit.waterflow.domain.utils.IdGenerator;
 import modelengine.fit.waterflow.domain.utils.UUIDUtil;
 
-import java.util.LinkedHashMap;
 import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-import java.util.concurrent.ConcurrentHashMap;
 import java.util.stream.Collectors;
 
 /**
@@ -77,44 +71,20 @@ private static <I> From<I> initFrom(String streamId, FlowContextRepo repo, FlowC
     }
 
     private static class ConditionFrom<I> extends From<I> {
-        private final Map<String, Map<String, Subscription<I>>> sessionSubscription = new ConcurrentHashMap<>();
-
         public ConditionFrom(String streamId, FlowContextRepo repo, FlowContextMessenger messenger, FlowLocks locks) {
             super(streamId, repo, messenger, locks);
         }
 
         @Override
         public void offer(List<FlowContext<I>> contexts) {
-            this.offerUserContexts(contexts);
-        }
-
-        private void offerUserContexts(List<FlowContext<I>> contexts) {
             this.getSubscriptions().forEach(subscription -> {
                 List<FlowContext<I>> matched = contexts.stream()
                         .filter(context -> subscription.getWhether().is(context.getData()))
-                        .peek(context -> {
-                            this.record(subscription, context);
-                        })
                         .collect(Collectors.toList());
                 matched.forEach(contexts::remove);
                 subscription.cache(matched);
             });
         }
 
-        private void record(Subscription<I> subscription, FlowContext<I> context) {
-            String sessionId = getSessionId(context);
-            if (sessionId == null) {
-                return;
-            }
-            this.sessionSubscription.putIfAbsent(sessionId, new LinkedHashMap<>());
-            Map<String, Subscription<I>> subscriptionMap = this.sessionSubscription.get(sessionId);
-            if (!subscriptionMap.containsKey(subscription.getId())) {
-                subscriptionMap.put(subscription.getId(), subscription);
-            }
-        }
-
-        private String getSessionId(FlowContext<I> context) {
-            return Optional.ofNullable(context.getSession()).map(IdGenerator::getId).orElse(null);
-        }
     }
 }