merge with feature/attr_tskv_version

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

@@ -38,10 +38,10 @@ public class CaffeineTbCacheTransaction<K extends Serializable, V extends Serial
     @Setter
     private boolean failed;
 
-    private final Map<Object, Object> pendingPuts = new LinkedHashMap<>();
+    private final Map<K, V> pendingPuts = new LinkedHashMap<>();
 
     @Override
-    public void putIfAbsent(K key, V value) {
+    public void put(K key, V value) {
         pendingPuts.put(key, value);
     }
 

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

@@ -17,6 +17,7 @@ package org.thingsboard.server.cache;
 
 import lombok.Getter;
 import lombok.RequiredArgsConstructor;
+import org.springframework.cache.Cache;
 import org.springframework.cache.CacheManager;
 
 import java.io.Serializable;
@@ -26,6 +27,7 @@ import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.locks.Lock;
@@ -34,17 +36,22 @@ import java.util.concurrent.locks.ReentrantLock;
 @RequiredArgsConstructor
 public abstract class CaffeineTbTransactionalCache<K extends Serializable, V extends Serializable> implements TbTransactionalCache<K, V> {
 
-    private final CacheManager cacheManager;
     @Getter
-    private final String cacheName;
-
-    private final Lock lock = new ReentrantLock();
+    protected final String cacheName;
+    protected final Cache cache;
+    protected final Lock lock = new ReentrantLock();
     private final Map<K, Set<UUID>> objectTransactions = new HashMap<>();
     private final Map<UUID, CaffeineTbCacheTransaction<K, V>> transactions = new HashMap<>();
 
+    public CaffeineTbTransactionalCache(CacheManager cacheManager, String cacheName) {
+        this.cacheName = cacheName;
+        this.cache = Optional.ofNullable(cacheManager.getCache(cacheName))
+                .orElseThrow(() -> new IllegalArgumentException("Cache '" + cacheName + "' is not configured"));
+    }
+
     @Override
     public TbCacheValueWrapper<V> get(K key) {
-        return SimpleTbCacheValueWrapper.wrap(cacheManager.getCache(cacheName).get(key));
+        return SimpleTbCacheValueWrapper.wrap(cache.get(key));
     }
 
     @Override
@@ -52,7 +59,7 @@ public abstract class CaffeineTbTransactionalCache<K extends Serializable, V ext
         lock.lock();
         try {
             failAllTransactionsByKey(key);
-            cacheManager.getCache(cacheName).put(key, value);
+            cache.put(key, value);
         } finally {
             lock.unlock();
         }
@@ -109,12 +116,12 @@ public abstract class CaffeineTbTransactionalCache<K extends Serializable, V ext
         return newTransaction(keys);
     }
 
-    void doPutIfAbsent(Object key, Object value) {
-        cacheManager.getCache(cacheName).putIfAbsent(key, value);
+    void doPutIfAbsent(K key, V value) {
+        cache.putIfAbsent(key, value);
     }
 
     void doEvict(K key) {
-        cacheManager.getCache(cacheName).evict(key);
+        cache.evict(key);
     }
 
     TbCacheTransaction<K, V> newTransaction(List<K> keys) {
@@ -132,7 +139,7 @@ public abstract class CaffeineTbTransactionalCache<K extends Serializable, V ext
         }
     }
 
-    public boolean commit(UUID trId, Map<Object, Object> pendingPuts) {
+    public boolean commit(UUID trId, Map<K, V> pendingPuts) {
         lock.lock();
         try {
             var tr = transactions.get(trId);
@@ -181,7 +188,7 @@ public abstract class CaffeineTbTransactionalCache<K extends Serializable, V ext
         }
     }
 
-    private void failAllTransactionsByKey(K key) {
+    protected void failAllTransactionsByKey(K key) {
         Set<UUID> transactionsIds = objectTransactions.get(key);
         if (transactionsIds != null) {
             for (UUID otherTrId : transactionsIds) {

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

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

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

@@ -103,10 +103,14 @@ 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);
+            put(key, value, connection);
         }
     }
 
+    public void put(K key, V value, RedisConnection connection) {
+        put(connection, key, value, RedisStringCommands.SetOption.UPSERT);
+    }
+
     @Override
     public void putIfAbsent(K key, V value) {
         try (var connection = connectionFactory.getConnection()) {

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

@@ -17,7 +17,7 @@ package org.thingsboard.server.cache;
 
 public interface TbCacheTransaction<K, V> {
 
-    void putIfAbsent(K key, V value);
+    void put(K key, V value);
 
     boolean commit();
 

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

@@ -68,7 +68,7 @@ public interface TbTransactionalCache<K extends Serializable, V extends Serializ
         try {
             V dbValue = dbCall.get();
             if (dbValue != null || cacheNullValue) {
-                cacheTransaction.putIfAbsent(key, dbValue);
+                cacheTransaction.put(key, dbValue);
                 cacheTransaction.commit();
                 return dbValue;
             } else {
@@ -104,7 +104,7 @@ public interface TbTransactionalCache<K extends Serializable, V extends Serializ
         try {
             R dbValue = dbCall.get();
             if (dbValue != null || cacheNullValue) {
-                cacheTransaction.putIfAbsent(key, dbValueToCacheValue.apply(dbValue));
+                cacheTransaction.put(key, dbValueToCacheValue.apply(dbValue));
                 cacheTransaction.commit();
                 return dbValue;
             } else {

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

@@ -15,23 +15,18 @@
  */
 package org.thingsboard.server.cache;
 
-import lombok.RequiredArgsConstructor;
 import org.springframework.cache.Cache;
 import org.springframework.cache.CacheManager;
 import org.thingsboard.server.common.data.HasVersion;
 import org.thingsboard.server.common.data.util.TbPair;
 
 import java.io.Serializable;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantLock;
 
-@RequiredArgsConstructor
-public abstract class VersionedCaffeineTbCache<K extends Serializable, V extends Serializable & HasVersion> implements VersionedTbCache<K, V> {
+public abstract class VersionedCaffeineTbCache<K extends Serializable, V extends Serializable & HasVersion> extends CaffeineTbTransactionalCache<K, V> implements VersionedTbCache<K, V> {
 
-    private final CacheManager cacheManager;
-    private final String cacheName;
-
-    private final Lock lock = new ReentrantLock();
+    public VersionedCaffeineTbCache(CacheManager cacheManager, String cacheName) {
+        super(cacheManager, cacheName);
+    }
 
     @Override
     public TbCacheValueWrapper<V> get(K key) {
@@ -56,7 +51,8 @@ public abstract class VersionedCaffeineTbCache<K extends Serializable, V extends
         try {
             TbPair<Long, V> versionValuePair = doGet(key);
             if (versionValuePair == null || version > versionValuePair.getFirst()) {
-                cacheManager.getCache(cacheName).put(key, TbPair.of(version, value));
+                failAllTransactionsByKey(key);
+                cache.put(key, wrapValue(value, version));
             }
         } finally {
             lock.unlock();
@@ -64,7 +60,7 @@ public abstract class VersionedCaffeineTbCache<K extends Serializable, V extends
     }
 
     private TbPair<Long, V> doGet(K key) {
-        Cache.ValueWrapper source = cacheManager.getCache(cacheName).get(key);
+        Cache.ValueWrapper source = cache.get(key);
         return source == null ? null : (TbPair<Long, V>) source.get();
     }
 
@@ -72,7 +68,8 @@ public abstract class VersionedCaffeineTbCache<K extends Serializable, V extends
     public void evict(K key) {
         lock.lock();
         try {
-            cacheManager.getCache(cacheName).evict(key);
+            failAllTransactionsByKey(key);
+            cache.evict(key);
         } finally {
             lock.unlock();
         }
@@ -83,11 +80,16 @@ public abstract class VersionedCaffeineTbCache<K extends Serializable, V extends
         if (version == null) {
             return;
         }
-        lock.lock();
-        try {
-            doPut(key, null, version);
-        } finally {
-            lock.unlock();
-        }
+        doPut(key, null, version);
     }
+
+    @Override
+    void doPutIfAbsent(K key, V value) {
+        cache.putIfAbsent(key, wrapValue(value, value != null ? value.getVersion() : 0));
+    }
+
+    private TbPair<Long, V> wrapValue(V value, Long version) {
+        return TbPair.of(version, value);
+    }
+
 }

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

@@ -28,8 +28,6 @@ import org.thingsboard.server.common.data.HasVersion;
 
 import java.io.Serializable;
 import java.util.Arrays;
-import java.util.Collection;
-import java.util.List;
 
 @Slf4j
 public abstract class VersionedRedisTbCache<K extends Serializable, V extends Serializable & HasVersion> extends RedisTbTransactionalCache<K, V> implements VersionedTbCache<K, V> {
@@ -47,7 +45,7 @@ public abstract class VersionedRedisTbCache<K extends Serializable, V extends Se
                 local newValueWithVersion = struct.pack(">I8", newVersion) .. newValue
                 redis.call('SET', key, newValueWithVersion, 'EX', expiration)
             end
-                        
+
             local function bytes_to_number(bytes)
                 local n = 0
                 for i = 1, 8 do
@@ -62,7 +60,7 @@ public abstract class VersionedRedisTbCache<K extends Serializable, V extends Se
             if currentVersionBytes and #currentVersionBytes == 8 then
                 local currentVersion = bytes_to_number(currentVersionBytes)
 
-                if newVersion > currentVersion or newVersion == 1 and currentVersion > 1 then
+                if newVersion > currentVersion then
                     setNewValue()
                 end
             else
@@ -96,7 +94,6 @@ public abstract class VersionedRedisTbCache<K extends Serializable, V extends Se
 
     @Override
     public void put(K key, V value) {
-        log.trace("put [{}][{}]", key, value);
         Long version;
         if (value == null) {
             version = 0L;
@@ -108,29 +105,48 @@ public abstract class VersionedRedisTbCache<K extends Serializable, V extends Se
         doPut(key, value, version, cacheTtl);
     }
 
+    @Override
+    public void put(K key, V value, RedisConnection connection) {
+        Long version;
+        if (value == null) {
+            version = 0L;
+        } else if (value.getVersion() != null) {
+            version = value.getVersion();
+        } else {
+            return;
+        }
+        byte[] rawKey = getRawKey(key);
+        doPut(rawKey, value, version, cacheTtl, connection);
+    }
+
     private void doPut(K key, V value, Long version, Expiration expiration) {
+        log.trace("put [{}][{}][{}]", key, value, version);
         if (version == null) {
             return;
         }
         final byte[] rawKey = getRawKey(key);
+        try (var connection = getConnection(rawKey)) {
+            doPut(rawKey, value, version, expiration, connection);
+        }
+    }
+
+    private void doPut(byte[] rawKey, V value, Long version, Expiration expiration, RedisConnection connection) {
         byte[] rawValue = getRawValue(value);
         byte[] rawVersion = StringRedisSerializer.UTF_8.serialize(String.valueOf(version));
         byte[] rawExpiration = StringRedisSerializer.UTF_8.serialize(String.valueOf(expiration.getExpirationTimeInSeconds()));
-        try (var connection = getConnection(rawKey)) {
+        try {
+            connection.scriptingCommands().evalSha(SET_VERSIONED_VALUE_SHA, ReturnType.VALUE, 1, rawKey, rawValue, rawVersion, rawExpiration);
+        } catch (InvalidDataAccessApiUsageException e) {
+            log.debug("loading LUA [{}]", connection.getNativeConnection());
+            String sha = connection.scriptingCommands().scriptLoad(SET_VERSIONED_VALUE_LUA_SCRIPT);
+            if (!Arrays.equals(SET_VERSIONED_VALUE_SHA, StringRedisSerializer.UTF_8.serialize(sha))) {
+                log.error("SHA for SET_VERSIONED_VALUE_LUA_SCRIPT wrong! Expected [{}], but actual [{}]", new String(SET_VERSIONED_VALUE_SHA), sha);
+            }
             try {
                 connection.scriptingCommands().evalSha(SET_VERSIONED_VALUE_SHA, ReturnType.VALUE, 1, rawKey, rawValue, rawVersion, rawExpiration);
-            } catch (InvalidDataAccessApiUsageException e) {
-                log.debug("loading LUA [{}]", connection.getNativeConnection());
-                String sha = connection.scriptingCommands().scriptLoad(SET_VERSIONED_VALUE_LUA_SCRIPT);
-                if (!Arrays.equals(SET_VERSIONED_VALUE_SHA, StringRedisSerializer.UTF_8.serialize(sha))) {
-                    log.error("SHA for SET_VERSIONED_VALUE_LUA_SCRIPT wrong! Expected [{}], but actual [{}]", new String(SET_VERSIONED_VALUE_SHA), sha);
-                }
-                try {
-                    connection.scriptingCommands().evalSha(SET_VERSIONED_VALUE_SHA, ReturnType.VALUE, 1, rawKey, rawValue, rawVersion, rawExpiration);
-                } catch (InvalidDataAccessApiUsageException ignored) {
-                    log.debug("Slowly executing eval instead of fast evalsha");
-                    connection.scriptingCommands().eval(SET_VERSIONED_VALUE_LUA_SCRIPT, ReturnType.VALUE, 1, rawKey, rawValue, rawVersion, rawExpiration);
-                }
+            } catch (InvalidDataAccessApiUsageException ignored) {
+                log.debug("Slowly executing eval instead of fast evalsha");
+                connection.scriptingCommands().eval(SET_VERSIONED_VALUE_LUA_SCRIPT, ReturnType.VALUE, 1, rawKey, rawValue, rawVersion, rawExpiration);
             }
         }
     }
@@ -145,28 +161,12 @@ public abstract class VersionedRedisTbCache<K extends Serializable, V extends Se
 
     @Override
     public void putIfAbsent(K key, V value) {
-        log.trace("putIfAbsent [{}][{}]", key, value);
         throw new NotImplementedException("putIfAbsent is not supported by versioned cache");
     }
 
-    @Override
-    public void evict(Collection<K> keys) {
-        throw new NotImplementedException("evict by many keys is not supported by versioned cache");
-    }
-
     @Override
     public void evictOrPut(K key, V value) {
         throw new NotImplementedException("evictOrPut is not supported by versioned cache");
     }
 
-    @Override
-    public TbCacheTransaction<K, V> newTransactionForKey(K key) {
-        throw new NotImplementedException("newTransactionForKey is not supported by versioned cache");
-    }
-
-    @Override
-    public TbCacheTransaction<K, V> newTransactionForKeys(List<K> keys) {
-        throw new NotImplementedException("newTransactionForKeys is not supported by versioned cache");
-    }
-
 }

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

@@ -18,14 +18,30 @@ package org.thingsboard.server.cache;
 import org.thingsboard.server.common.data.HasVersion;
 
 import java.io.Serializable;
+import java.util.Collection;
+import java.util.Optional;
+import java.util.function.Supplier;
 
-public interface VersionedTbCache<K extends Serializable, V extends Serializable & HasVersion> {
+public interface VersionedTbCache<K extends Serializable, V extends Serializable & HasVersion> extends TbTransactionalCache<K, V> {
 
     TbCacheValueWrapper<V> get(K key);
 
+    default V get(K key, Supplier<V> supplier) {
+        return Optional.ofNullable(get(key))
+                .map(TbCacheValueWrapper::get)
+                .orElseGet(() -> {
+                    V value = supplier.get();
+                    put(key, value);
+                    return value;
+                });
+    }
+
     void put(K key, V value);
 
     void evict(K key);
 
+    void evict(Collection<K> keys);
+
     void evict(K key, Long version);
+
 }

dao/src/test/resources/application-test.properties

@@ -63,8 +63,8 @@ cache.specs.attributes.maxSize=100000
 cache.specs.tsLatest.timeToLiveInMinutes=1440
 cache.specs.tsLatest.maxSize=100000
 
-cache.specs.tokensOutdatageTime.timeToLiveInMinutes=1440
-cache.specs.tokensOutdatageTime.maxSize=100000
+cache.specs.userSessionsInvalidation.timeToLiveInMinutes=1440
+cache.specs.userSessionsInvalidation.maxSize=10000
 
 cache.specs.otaPackages.timeToLiveInMinutes=1440
 cache.specs.otaPackages.maxSize=100000
@@ -93,6 +93,15 @@ cache.specs.resourceInfo.maxSize=10000
 cache.specs.alarmTypes.timeToLiveInMinutes=60
 cache.specs.alarmTypes.maxSize=10000
 
+cache.specs.userSettings.timeToLiveInMinutes=1440
+cache.specs.userSettings.maxSize=10000
+
+cache.specs.mobileAppSettings.timeToLiveInMinutes=1440
+cache.specs.mobileAppSettings.maxSize=10000
+
+cache.specs.mobileSecretKey.timeToLiveInMinutes=1440
+cache.specs.mobileSecretKey.maxSize=10000
+
 redis.connection.host=localhost
 redis.connection.port=6379
 redis.connection.db=0