DeviceActorMessageProcessor: introduced LinkedHashMapRemoveEldest with BiConsume…

…r instead many ensureSessionsCapacity() that spreaded over the class. test added

@@ -24,6 +24,7 @@ import com.google.protobuf.InvalidProtocolBufferException;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.collections.CollectionUtils;
 import org.thingsboard.common.util.JacksonUtil;
+import org.thingsboard.common.util.LinkedHashMapRemoveEldest;
 import org.thingsboard.rule.engine.api.RpcError;
 import org.thingsboard.rule.engine.api.msg.DeviceAttributesEventNotificationMsg;
 import org.thingsboard.rule.engine.api.msg.DeviceCredentialsUpdateNotificationMsg;
@@ -112,7 +113,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
 
     final TenantId tenantId;
     final DeviceId deviceId;
-    private final LinkedHashMap<UUID, SessionInfoMetaData> sessions;
+    final LinkedHashMapRemoveEldest<UUID, SessionInfoMetaData> sessions;
     private final Map<UUID, SessionInfo> attributeSubscriptions;
     private final Map<UUID, SessionInfo> rpcSubscriptions;
     private final Map<Integer, ToDeviceRpcRequestMetadata> toDeviceRpcPendingMap;
@@ -127,16 +128,16 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
         super(systemContext);
         this.tenantId = tenantId;
         this.deviceId = deviceId;
-        this.sessions = new LinkedHashMap<>();
         this.attributeSubscriptions = new HashMap<>();
         this.rpcSubscriptions = new HashMap<>();
         this.toDeviceRpcPendingMap = new HashMap<>();
+        this.sessions = new LinkedHashMapRemoveEldest<>(systemContext.getMaxConcurrentSessionsPerDevice(), this::notifyTransportAboutClosedSession);
         if (initAttributes()) {
             restoreSessions();
         }
     }
 
-    private boolean initAttributes() {
+    boolean initAttributes() {
         Device device = systemContext.getDeviceService().findDeviceById(tenantId, deviceId);
         if (device != null) {
             this.deviceName = device.getName();
@@ -562,8 +563,6 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
             }
             log.info("[{}] Processing new session [{}]. Current sessions size {}", deviceId, sessionId, sessions.size());
 
-            ensureSessionsCapacity();
-
             sessions.put(sessionId, new SessionInfoMetaData(new SessionInfo(SessionType.ASYNC, sessionInfo.getNodeId())));
             if (sessions.size() == 1) {
                 reportSessionOpen();
@@ -582,24 +581,10 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
         }
     }
 
-    private void ensureSessionsCapacity() {
-        while (sessions.size() >= systemContext.getMaxConcurrentSessionsPerDevice()) {
-            Optional<UUID> sessionIdToRemove = sessions.keySet().stream().findFirst();
-            if (sessionIdToRemove.isPresent()) {
-                notifyTransportAboutClosedSession(sessionIdToRemove.get(), sessions.remove(sessionIdToRemove.get()), "max concurrent sessions limit reached per device!");
-            } else {
-                log.warn("[{}] Can't remove session because find first returns null", deviceId);
-            }
-        }
-        log.debug("[{}] sessions size after clean up {}", deviceId, sessions.size());
-    }
-
     private void handleSessionActivity(TbActorCtx context, SessionInfoProto sessionInfoProto, SubscriptionInfoProto subscriptionInfo) {
         UUID sessionId = getSessionId(sessionInfoProto);
         Objects.requireNonNull(sessionId);
 
-        ensureSessionsCapacity();
-
         SessionInfoMetaData sessionMD = sessions.computeIfAbsent(sessionId,
                 id -> new SessionInfoMetaData(new SessionInfo(SessionType.ASYNC, sessionInfoProto.getNodeId()), subscriptionInfo.getLastActivityTime()));
 
@@ -777,7 +762,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
         return builder.build();
     }
 
-    private void restoreSessions() {
+    void restoreSessions() {
         log.debug("[{}] Restoring sessions from cache", deviceId);
         DeviceSessionsCacheEntry sessionsDump = null;
         try {
@@ -809,13 +794,9 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
             log.debug("[{}] Restored session: {}", deviceId, sessionMD);
         }
         log.debug("[{}] Restored sessions: {}, rpc subscriptions: {}, attribute subscriptions: {}", deviceId, sessions.size(), rpcSubscriptions.size(), attributeSubscriptions.size());
-
-        ensureSessionsCapacity();
     }
 
     private void dumpSessions() {
-        ensureSessionsCapacity();
-
         log.debug("[{}] Dumping sessions: {}, rpc subscriptions: {}, attribute subscriptions: {} to cache", deviceId, sessions.size(), rpcSubscriptions.size(), attributeSubscriptions.size());
         List<SessionSubscriptionInfoProto> sessionsList = new ArrayList<>(sessions.size());
         sessions.forEach((uuid, sessionMD) -> {

+/**
+ * Copyright © 2016-2021 The Thingsboard Authors
+ * <p>
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.actors.device;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.thingsboard.common.util.LinkedHashMapRemoveEldest;
+import org.thingsboard.server.actors.ActorSystemContext;
+import org.thingsboard.server.common.data.id.DeviceId;
+import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.dao.device.DeviceService;
+
+import static org.hamcrest.CoreMatchers.instanceOf;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.CoreMatchers.notNullValue;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.mockito.BDDMockito.willReturn;
+import static org.mockito.Mockito.mock;
+
+public class DeviceActorMessageProcessorTest {
+
+    public static final long MAX_CONCURRENT_SESSIONS_PER_DEVICE = 10L;
+    ActorSystemContext systemContext;
+    DeviceService deviceService;
+    TenantId tenantId = TenantId.SYS_TENANT_ID;
+    DeviceId deviceId = DeviceId.fromString("78bf9b26-74ef-4af2-9cfb-ad6cf24ad2ec");
+
+    DeviceActorMessageProcessor processor;
+
+    @Before
+    public void setUp() {
+        systemContext = mock(ActorSystemContext.class);
+        deviceService = mock(DeviceService.class);
+        willReturn(MAX_CONCURRENT_SESSIONS_PER_DEVICE).given(systemContext).getMaxConcurrentSessionsPerDevice();
+        willReturn(deviceService).given(systemContext).getDeviceService();
+        processor = new DeviceActorMessageProcessor(systemContext, tenantId, deviceId);
+    }
+
+    @Test
+    public void givenSystemContext_whenNewInstance_thenVerifySessionMapMaxSize() {
+        assertThat(processor.sessions, instanceOf(LinkedHashMapRemoveEldest.class));
+        assertThat(processor.sessions.getMaxEntries(), is(MAX_CONCURRENT_SESSIONS_PER_DEVICE));
+        assertThat(processor.sessions.getRemovalConsumer(), notNullValue());
+    }
+}
\ No newline at end of file