From b13b810e4eb812042fde9be16e6db3d0b858e9cc Mon Sep 17 00:00:00 2001
From: vparomskiy <paromskiy@gmail.com>
Date: Mon, 11 Jun 2018 18:13:02 +0300
Subject: [PATCH] use threadsafe collection during subscription

---
 .../telemetry/DefaultTelemetrySubscriptionService.java | 10 +++++-----
 1 file changed, 5 insertions(+), 5 deletions(-)

diff --git a/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java
index dbd58d540c..ef1606a3a3 100644
--- a/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java
+++ b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java
@@ -60,7 +60,6 @@ import javax.annotation.PostConstruct;
 import javax.annotation.PreDestroy;
 import java.util.ArrayList;
 import java.util.Collections;
-import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
@@ -68,6 +67,7 @@ import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 import java.util.TreeMap;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.function.Consumer;
@@ -120,8 +120,8 @@ public class DefaultTelemetrySubscriptionService implements TelemetrySubscriptio
         }
     }
 
-    private final Map<EntityId, Set<Subscription>> subscriptionsByEntityId = new HashMap<>();
-    private final Map<String, Map<Integer, Subscription>> subscriptionsByWsSessionId = new HashMap<>();
+    private final Map<EntityId, Set<Subscription>> subscriptionsByEntityId = new ConcurrentHashMap<>();
+    private final Map<String, Map<Integer, Subscription>> subscriptionsByWsSessionId = new ConcurrentHashMap<>();
 
     @Override
     public void addLocalWsSubscription(String sessionId, EntityId entityId, SubscriptionState sub) {
@@ -453,9 +453,9 @@ public class DefaultTelemetrySubscriptionService implements TelemetrySubscriptio
     }
 
     private void registerSubscription(String sessionId, EntityId entityId, Subscription subscription) {
-        Set<Subscription> deviceSubscriptions = subscriptionsByEntityId.computeIfAbsent(entityId, k -> new HashSet<>());
+        Set<Subscription> deviceSubscriptions = subscriptionsByEntityId.computeIfAbsent(entityId, k -> ConcurrentHashMap.newKeySet());
         deviceSubscriptions.add(subscription);
-        Map<Integer, Subscription> sessionSubscriptions = subscriptionsByWsSessionId.computeIfAbsent(sessionId, k -> new HashMap<>());
+        Map<Integer, Subscription> sessionSubscriptions = subscriptionsByWsSessionId.computeIfAbsent(sessionId, k -> new ConcurrentHashMap<>());
         sessionSubscriptions.put(subscription.getSubscriptionId(), subscription);
     }