Session Cache

application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java

@@ -884,10 +884,10 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
             return;
         }
         log.debug("[{}] Restoring sessions from cache", deviceId);
-        DeviceSessionsCacheEntry sessionsDump = null;
+        DeviceSessionsCacheEntry sessionsDump;
         try {
-            sessionsDump = DeviceSessionsCacheEntry.parseFrom(systemContext.getDeviceSessionCacheService().get(deviceId));
-        } catch (InvalidProtocolBufferException e) {
+            sessionsDump = systemContext.getDeviceSessionCacheService().get(deviceId);
+        } catch (Exception e) {
             log.warn("[{}] Failed to decode device sessions from cache", deviceId);
             return;
         }
@@ -942,7 +942,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
         });
         systemContext.getDeviceSessionCacheService()
                 .put(deviceId, DeviceSessionsCacheEntry.newBuilder()
-                        .addAllSessions(sessionsList).build().toByteArray());
+                        .addAllSessions(sessionsList).build());
     }
 
     void init(TbActorCtx ctx) {

application/src/main/java/org/thingsboard/server/service/session/DefaultDeviceSessionCacheService.java

@@ -15,14 +15,18 @@
  */
 package org.thingsboard.server.service.session;
 
+import com.google.protobuf.InvalidProtocolBufferException;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.cache.annotation.CachePut;
 import org.springframework.cache.annotation.Cacheable;
 import org.springframework.stereotype.Service;
+import org.thingsboard.server.cache.TbTransactionalCache;
 import org.thingsboard.server.common.data.id.DeviceId;
 import org.thingsboard.server.gen.transport.TransportProtos.DeviceSessionsCacheEntry;
 import org.thingsboard.server.queue.util.TbCoreComponent;
 
+import java.io.Serializable;
 import java.util.Collections;
 
 import static org.thingsboard.server.common.data.CacheConstants.SESSIONS_CACHE;
@@ -35,17 +39,20 @@ import static org.thingsboard.server.common.data.CacheConstants.SESSIONS_CACHE;
 @Slf4j
 public class DefaultDeviceSessionCacheService implements DeviceSessionCacheService {
 
+    @Autowired
+    protected TbTransactionalCache<DeviceId, DeviceSessionsCacheEntry> cache;
+
     @Override
-    @Cacheable(cacheNames = SESSIONS_CACHE, key = "#deviceId.toString()")
-    public byte[] get(DeviceId deviceId) {
+    public DeviceSessionsCacheEntry get(DeviceId deviceId) {
         log.debug("[{}] Fetching session data from cache", deviceId);
-        return DeviceSessionsCacheEntry.newBuilder().addAllSessions(Collections.emptyList()).build().toByteArray();
+        return cache.getAndPutInTransaction(deviceId, () ->
+                DeviceSessionsCacheEntry.newBuilder().addAllSessions(Collections.emptyList()).build(), false);
     }
 
     @Override
-    @CachePut(cacheNames = SESSIONS_CACHE, key = "#deviceId.toString()")
-    public byte[] put(DeviceId deviceId, byte[] sessions) {
+    public DeviceSessionsCacheEntry put(DeviceId deviceId, DeviceSessionsCacheEntry sessions) {
         log.debug("[{}] Pushing session data to cache: {}", deviceId, sessions);
+        cache.putIfAbsent(deviceId, sessions);
         return sessions;
     }
 }

application/src/main/java/org/thingsboard/server/service/session/DeviceSessionCacheService.java

@@ -23,8 +23,8 @@ import org.thingsboard.server.gen.transport.TransportProtos.DeviceSessionsCacheE
  */
 public interface DeviceSessionCacheService {
 
-    byte[] get(DeviceId deviceId);
+    DeviceSessionsCacheEntry get(DeviceId deviceId);
 
-    byte[] put(DeviceId deviceId, byte[] sessions);
+    DeviceSessionsCacheEntry put(DeviceId deviceId, DeviceSessionsCacheEntry sessions);
 
 }

application/src/main/java/org/thingsboard/server/service/session/SessionCaffeineCache.java

@@ -0,0 +1,37 @@
+/**
+ * Copyright © 2016-2022 The Thingsboard Authors
+ *
+ * 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
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.service.session;
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.cache.CacheManager;
+import org.springframework.stereotype.Service;
+import org.thingsboard.server.common.data.CacheConstants;
+import org.thingsboard.server.common.data.DeviceInfo;
+import org.thingsboard.server.common.data.asset.Asset;
+import org.thingsboard.server.common.data.id.DeviceId;
+import org.thingsboard.server.dao.asset.AssetCacheKey;
+import org.thingsboard.server.dao.cache.CaffeineTbTransactionalCache;
+import org.thingsboard.server.gen.transport.TransportProtos;
+
+@ConditionalOnProperty(prefix = "cache", value = "type", havingValue = "caffeine", matchIfMissing = true)
+@Service("SessionCache")
+public class SessionCaffeineCache extends CaffeineTbTransactionalCache<DeviceId, TransportProtos.DeviceSessionsCacheEntry> {
+
+    public SessionCaffeineCache(CacheManager cacheManager) {
+        super(cacheManager, CacheConstants.SESSIONS_CACHE);
+    }
+
+}

application/src/main/java/org/thingsboard/server/service/session/SessionRedisCache.java

@@ -0,0 +1,55 @@
+/**
+ * Copyright © 2016-2022 The Thingsboard Authors
+ *
+ * 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
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.service.session;
+
+import com.google.protobuf.InvalidProtocolBufferException;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.serializer.RedisSerializer;
+import org.springframework.data.redis.serializer.SerializationException;
+import org.springframework.stereotype.Service;
+import org.thingsboard.server.cache.CacheSpecsMap;
+import org.thingsboard.server.cache.TBRedisCacheConfiguration;
+import org.thingsboard.server.common.data.CacheConstants;
+import org.thingsboard.server.common.data.asset.Asset;
+import org.thingsboard.server.common.data.id.DeviceId;
+import org.thingsboard.server.dao.asset.AssetCacheKey;
+import org.thingsboard.server.dao.cache.RedisTbTransactionalCache;
+import org.thingsboard.server.dao.cache.TbRedisSerializer;
+import org.thingsboard.server.gen.transport.TransportProtos;
+
+@ConditionalOnProperty(prefix = "cache", value = "type", havingValue = "redis")
+@Service("SessionCache")
+public class SessionRedisCache extends RedisTbTransactionalCache<DeviceId, TransportProtos.DeviceSessionsCacheEntry> {
+
+    public SessionRedisCache(TBRedisCacheConfiguration configuration, CacheSpecsMap cacheSpecsMap, RedisConnectionFactory connectionFactory) {
+        super(CacheConstants.ASSET_CACHE, cacheSpecsMap, connectionFactory, configuration, new RedisSerializer<>() {
+            @Override
+            public byte[] serialize(TransportProtos.DeviceSessionsCacheEntry deviceSessionsCacheEntry) throws SerializationException {
+                return deviceSessionsCacheEntry.toByteArray();
+            }
+
+            @Override
+            public TransportProtos.DeviceSessionsCacheEntry deserialize(byte[] bytes) throws SerializationException {
+                try {
+                    return TransportProtos.DeviceSessionsCacheEntry.parseFrom(bytes);
+                } catch (InvalidProtocolBufferException e) {
+                    throw new RuntimeException("Failed to deserialize session cache entry");
+                }
+            }
+        });
+    }
+}

common/cache/src/main/java/org/thingsboard/server/cache/TbTransactionalCache.java

@@ -27,6 +27,8 @@ public interface TbTransactionalCache<K extends Serializable, V extends Serializ
 
     TbCacheValueWrapper<V> get(K key);
 
+    void put(K key, V value);
+
     void putIfAbsent(K key, V value);
 
     void evict(K key);

dao/src/main/java/org/thingsboard/server/dao/cache/CaffeineTbTransactionalCache.java

@@ -50,6 +50,17 @@ public abstract class CaffeineTbTransactionalCache<K extends Serializable, V ext
         return SimpleTbCacheValueWrapper.wrap(cacheManager.getCache(cacheName).get(key));
     }
 
+    @Override
+    public void put(K key, V value) {
+        lock.lock();
+        try {
+            failAllTransactionsByKey(key);
+            cacheManager.getCache(cacheName).put(key, value);
+        } finally {
+            lock.unlock();
+        }
+    }
+
     @Override
     public void putIfAbsent(K key, V value) {
         lock.lock();

dao/src/main/java/org/thingsboard/server/dao/cache/RedisTbCacheTransaction.java

@@ -18,6 +18,7 @@ package org.thingsboard.server.dao.cache;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.data.redis.connection.RedisConnection;
+import org.springframework.data.redis.connection.RedisStringCommands;
 import org.thingsboard.server.cache.TbCacheTransaction;
 
 import java.io.Serializable;
@@ -32,7 +33,7 @@ public class RedisTbCacheTransaction<K extends Serializable, V extends Serializa
 
     @Override
     public void putIfAbsent(K key, V value) {
-        cache.putIfAbsent(connection, key, value);
+        cache.put(connection, key, value, RedisStringCommands.SetOption.UPSERT);
     }
 
     @Override

dao/src/main/java/org/thingsboard/server/dao/cache/RedisTbTransactionalCache.java

@@ -16,7 +16,6 @@
 package org.thingsboard.server.dao.cache;
 
 import lombok.Getter;
-import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.cache.support.NullValue;
 import org.springframework.data.redis.connection.RedisConnection;
@@ -83,10 +82,17 @@ public abstract class RedisTbTransactionalCache<K extends Serializable, V extend
         }
     }
 
+    @Override
+    public void put(K key, V value) {
+        try (var connection = connectionFactory.getConnection()) {
+            put(connection, key, value, RedisStringCommands.SetOption.UPSERT);
+        }
+    }
+
     @Override
     public void putIfAbsent(K key, V value) {
         try (var connection = connectionFactory.getConnection()) {
-            putIfAbsent(connection, key, value);
+            put(connection, key, value, RedisStringCommands.SetOption.SET_IF_ABSENT);
         }
     }
 
@@ -170,10 +176,10 @@ public abstract class RedisTbTransactionalCache<K extends Serializable, V extend
         }
     }
 
-    public void putIfAbsent(RedisConnection connection, K key, V value) {
+    public void put(RedisConnection connection, K key, V value, RedisStringCommands.SetOption setOption) {
         byte[] rawKey = getRawKey(key);
         byte[] rawValue = getRawValue(value);
-        connection.set(rawKey, rawValue, cacheTtl, RedisStringCommands.SetOption.SET_IF_ABSENT);
+        connection.set(rawKey, rawValue, cacheTtl, setOption);
     }
 
 }