Commit b13b810e4eb812042fde9be16e6db3d0b858e9cc

Authored by vparomskiy
1 parent 5ba6b67c

use threadsafe collection during subscription

@@ -60,7 +60,6 @@ import javax.annotation.PostConstruct; @@ -60,7 +60,6 @@ import javax.annotation.PostConstruct;
60 import javax.annotation.PreDestroy; 60 import javax.annotation.PreDestroy;
61 import java.util.ArrayList; 61 import java.util.ArrayList;
62 import java.util.Collections; 62 import java.util.Collections;
63 -import java.util.HashMap;  
64 import java.util.HashSet; 63 import java.util.HashSet;
65 import java.util.Iterator; 64 import java.util.Iterator;
66 import java.util.List; 65 import java.util.List;
@@ -68,6 +67,7 @@ import java.util.Map; @@ -68,6 +67,7 @@ import java.util.Map;
68 import java.util.Optional; 67 import java.util.Optional;
69 import java.util.Set; 68 import java.util.Set;
70 import java.util.TreeMap; 69 import java.util.TreeMap;
  70 +import java.util.concurrent.ConcurrentHashMap;
71 import java.util.concurrent.ExecutorService; 71 import java.util.concurrent.ExecutorService;
72 import java.util.concurrent.Executors; 72 import java.util.concurrent.Executors;
73 import java.util.function.Consumer; 73 import java.util.function.Consumer;
@@ -120,8 +120,8 @@ public class DefaultTelemetrySubscriptionService implements TelemetrySubscriptio @@ -120,8 +120,8 @@ public class DefaultTelemetrySubscriptionService implements TelemetrySubscriptio
120 } 120 }
121 } 121 }
122 122
123 - private final Map<EntityId, Set<Subscription>> subscriptionsByEntityId = new HashMap<>();  
124 - private final Map<String, Map<Integer, Subscription>> subscriptionsByWsSessionId = new HashMap<>(); 123 + private final Map<EntityId, Set<Subscription>> subscriptionsByEntityId = new ConcurrentHashMap<>();
  124 + private final Map<String, Map<Integer, Subscription>> subscriptionsByWsSessionId = new ConcurrentHashMap<>();
125 125
126 @Override 126 @Override
127 public void addLocalWsSubscription(String sessionId, EntityId entityId, SubscriptionState sub) { 127 public void addLocalWsSubscription(String sessionId, EntityId entityId, SubscriptionState sub) {
@@ -453,9 +453,9 @@ public class DefaultTelemetrySubscriptionService implements TelemetrySubscriptio @@ -453,9 +453,9 @@ public class DefaultTelemetrySubscriptionService implements TelemetrySubscriptio
453 } 453 }
454 454
455 private void registerSubscription(String sessionId, EntityId entityId, Subscription subscription) { 455 private void registerSubscription(String sessionId, EntityId entityId, Subscription subscription) {
456 - Set<Subscription> deviceSubscriptions = subscriptionsByEntityId.computeIfAbsent(entityId, k -> new HashSet<>()); 456 + Set<Subscription> deviceSubscriptions = subscriptionsByEntityId.computeIfAbsent(entityId, k -> ConcurrentHashMap.newKeySet());
457 deviceSubscriptions.add(subscription); 457 deviceSubscriptions.add(subscription);
458 - Map<Integer, Subscription> sessionSubscriptions = subscriptionsByWsSessionId.computeIfAbsent(sessionId, k -> new HashMap<>()); 458 + Map<Integer, Subscription> sessionSubscriptions = subscriptionsByWsSessionId.computeIfAbsent(sessionId, k -> new ConcurrentHashMap<>());
459 sessionSubscriptions.put(subscription.getSubscriptionId(), subscription); 459 sessionSubscriptions.put(subscription.getSubscriptionId(), subscription);
460 } 460 }
461 461