Andrii Shvaika
4 years ago
22 changed files with 776 additions and 132 deletions
@ -0,0 +1,118 @@ |
|||
package org.thingsboard.server.cache; |
|||
|
|||
import lombok.RequiredArgsConstructor; |
|||
|
|||
import java.io.Serializable; |
|||
import java.util.HashMap; |
|||
import java.util.HashSet; |
|||
import java.util.List; |
|||
import java.util.Map; |
|||
import java.util.Set; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.locks.Lock; |
|||
import java.util.concurrent.locks.ReentrantLock; |
|||
|
|||
@RequiredArgsConstructor |
|||
class CaffeineCacheTransactionStorage { |
|||
|
|||
private final String cacheName; |
|||
private final CaffeineTbTransactionalCacheManager cache; |
|||
private final Lock lock = new ReentrantLock(); |
|||
private final Map<Object, Set<UUID>> objectTransactions = new HashMap<>(); |
|||
private final Map<UUID, CaffeineTbCacheTransaction> transactions = new HashMap<>(); |
|||
|
|||
|
|||
<K extends Serializable> TbCacheTransaction newTransaction(List<K> keys) { |
|||
lock.lock(); |
|||
try { |
|||
var transaction = new CaffeineTbCacheTransaction(this, keys); |
|||
var transactionId = transaction.getId(); |
|||
for (K key : keys) { |
|||
objectTransactions.computeIfAbsent(key, k -> new HashSet<>()).add(transactionId); |
|||
} |
|||
transactions.put(transactionId, transaction); |
|||
return transaction; |
|||
} finally { |
|||
lock.unlock(); |
|||
} |
|||
} |
|||
|
|||
<K extends Serializable, V extends Serializable> void putIfAbsent(K key, V value) { |
|||
lock.lock(); |
|||
try { |
|||
failAllTransactionsByKey(key); |
|||
cache.doPutIfAbsent(cacheName, key, value); |
|||
} finally { |
|||
lock.unlock(); |
|||
} |
|||
} |
|||
|
|||
public <K extends Serializable> void evict(K key) { |
|||
lock.lock(); |
|||
try { |
|||
failAllTransactionsByKey(key); |
|||
cache.doEvict(cacheName, key); |
|||
} finally { |
|||
lock.unlock(); |
|||
} |
|||
} |
|||
|
|||
public boolean commit(UUID trId, Map<Object, Object> pendingPuts) { |
|||
lock.lock(); |
|||
try { |
|||
var tr = transactions.get(trId); |
|||
var success = !tr.isFailed(); |
|||
if (success) { |
|||
for (Object key : tr.getKeys()) { |
|||
Set<UUID> otherTransactions = objectTransactions.get(key); |
|||
if (otherTransactions != null) { |
|||
for (UUID otherTrId : otherTransactions) { |
|||
if (trId == null || !trId.equals(otherTrId)) { |
|||
transactions.get(otherTrId).setFailed(true); |
|||
} |
|||
} |
|||
} |
|||
} |
|||
pendingPuts.forEach((k, v) -> cache.doPutIfAbsent(cacheName, k, v)); |
|||
} |
|||
removeTransaction(trId); |
|||
return success; |
|||
} finally { |
|||
lock.unlock(); |
|||
} |
|||
} |
|||
|
|||
void rollback(UUID id) { |
|||
lock.lock(); |
|||
try { |
|||
removeTransaction(id); |
|||
} finally { |
|||
lock.unlock(); |
|||
} |
|||
} |
|||
|
|||
private void removeTransaction(UUID id) { |
|||
CaffeineTbCacheTransaction transaction = transactions.remove(id); |
|||
if (transaction != null) { |
|||
for (var key : transaction.getKeys()) { |
|||
Set<UUID> transactions = objectTransactions.get(key); |
|||
if (transactions != null) { |
|||
transactions.remove(id); |
|||
if (transactions.isEmpty()) { |
|||
objectTransactions.remove(key); |
|||
} |
|||
} |
|||
} |
|||
} |
|||
} |
|||
|
|||
private <K extends Serializable> void failAllTransactionsByKey(K key) { |
|||
Set<UUID> transactionsIds = objectTransactions.get(key); |
|||
if (transactionsIds != null) { |
|||
for (UUID otherTrId : transactionsIds) { |
|||
transactions.get(otherTrId).setFailed(true); |
|||
} |
|||
} |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,61 @@ |
|||
package org.thingsboard.server.cache; |
|||
|
|||
import com.google.common.util.concurrent.FutureCallback; |
|||
import com.google.common.util.concurrent.Futures; |
|||
import com.google.common.util.concurrent.ListenableFuture; |
|||
import lombok.Getter; |
|||
import lombok.RequiredArgsConstructor; |
|||
import lombok.Setter; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.checkerframework.checker.nullness.qual.Nullable; |
|||
|
|||
import java.util.LinkedHashMap; |
|||
import java.util.List; |
|||
import java.util.Map; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.Executor; |
|||
|
|||
@Slf4j |
|||
@RequiredArgsConstructor |
|||
public class CaffeineTbCacheTransaction implements TbCacheTransaction { |
|||
@Getter |
|||
private final UUID id = UUID.randomUUID(); |
|||
private final CaffeineCacheTransactionStorage cache; |
|||
@Getter |
|||
private final List<?> keys; |
|||
@Getter @Setter |
|||
private boolean failed; |
|||
|
|||
private final Map<Object, Object> pendingPuts = new LinkedHashMap<>(); |
|||
|
|||
@Override |
|||
public <K, V> void putIfAbsent(K key, V value) { |
|||
pendingPuts.put(key, value); |
|||
} |
|||
|
|||
@Override |
|||
public boolean commit() { |
|||
return cache.commit(id, pendingPuts); |
|||
} |
|||
|
|||
@Override |
|||
public void rollback() { |
|||
cache.rollback(id); |
|||
} |
|||
|
|||
@Override |
|||
public <T> void rollBackOnFailure(ListenableFuture<T> future, Executor executor) { |
|||
Futures.addCallback(future, new FutureCallback<T>() { |
|||
@Override |
|||
public void onSuccess(@Nullable T result) { |
|||
} |
|||
|
|||
@Override |
|||
public void onFailure(Throwable t) { |
|||
log.trace("[{}] Rollback transaction due to error", id, t); |
|||
rollback(); |
|||
} |
|||
}, executor); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,59 @@ |
|||
package org.thingsboard.server.cache; |
|||
|
|||
import lombok.RequiredArgsConstructor; |
|||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; |
|||
import org.springframework.cache.Cache; |
|||
import org.springframework.cache.CacheManager; |
|||
import org.springframework.stereotype.Service; |
|||
|
|||
import java.io.Serializable; |
|||
import java.util.Collections; |
|||
import java.util.List; |
|||
import java.util.concurrent.ConcurrentHashMap; |
|||
import java.util.concurrent.ConcurrentMap; |
|||
|
|||
@ConditionalOnProperty(prefix = "cache", value = "type", havingValue = "caffeine", matchIfMissing = true) |
|||
@Service |
|||
@RequiredArgsConstructor |
|||
public class CaffeineTbTransactionalCacheManager implements TbTransactionalCache { |
|||
|
|||
private final CacheManager cacheManager; |
|||
private final ConcurrentMap<String, CaffeineCacheTransactionStorage> caches = new ConcurrentHashMap<>(); |
|||
|
|||
@Override |
|||
public <K extends Serializable> Cache.ValueWrapper get(String cacheName, K key) { |
|||
return cacheManager.getCache(cacheName).get(key); |
|||
} |
|||
|
|||
@Override |
|||
public <K extends Serializable, V extends Serializable> void putIfAbsent(String cacheName, K key, V value) { |
|||
getCache(cacheName).putIfAbsent(key, value); |
|||
} |
|||
|
|||
@Override |
|||
public <K extends Serializable> void evict(String cacheName, K key) { |
|||
getCache(cacheName).evict(key); |
|||
} |
|||
|
|||
@Override |
|||
public <K extends Serializable> TbCacheTransaction newTransactionForKey(String cacheName, K key) { |
|||
return getCache(cacheName).newTransaction(Collections.singletonList(key)); |
|||
} |
|||
|
|||
@Override |
|||
public <K extends Serializable> TbCacheTransaction newTransactionForKeys(String cacheName, List<K> keys) { |
|||
return getCache(cacheName).newTransaction(keys); |
|||
} |
|||
|
|||
private CaffeineCacheTransactionStorage getCache(String cacheName) { |
|||
return caches.computeIfAbsent(cacheName, cn -> new CaffeineCacheTransactionStorage(cacheName, this)); |
|||
} |
|||
|
|||
<K extends Serializable, V extends Serializable> void doPutIfAbsent(String cacheName, Object key, Object value) { |
|||
cacheManager.getCache(cacheName).putIfAbsent(key, value); |
|||
} |
|||
|
|||
<K extends Serializable> void doEvict(String cacheName, K key) { |
|||
cacheManager.getCache(cacheName).evict(key); |
|||
} |
|||
} |
|||
@ -0,0 +1,42 @@ |
|||
package org.thingsboard.server.cache; |
|||
|
|||
import lombok.RequiredArgsConstructor; |
|||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; |
|||
import org.springframework.cache.Cache; |
|||
import org.springframework.cache.CacheManager; |
|||
import org.springframework.stereotype.Service; |
|||
|
|||
import java.io.Serializable; |
|||
import java.util.List; |
|||
|
|||
@ConditionalOnProperty(prefix = "cache", value = "type", havingValue = "redis") |
|||
@Service |
|||
@RequiredArgsConstructor |
|||
public class RedisTbTransactionalCacheManager implements TbTransactionalCache { |
|||
|
|||
private final CacheManager cacheManager; |
|||
|
|||
@Override |
|||
public <K extends Serializable> Cache.ValueWrapper get(String cacheName, K key) { |
|||
return cacheManager.getCache(cacheName).get(key); |
|||
} |
|||
|
|||
@Override |
|||
public <K extends Serializable, V extends Serializable> void putIfAbsent(String cacheName, K key, V value) { |
|||
} |
|||
|
|||
@Override |
|||
public <K extends Serializable> void evict(String cacheName, K key) { |
|||
} |
|||
|
|||
@Override |
|||
public <K extends Serializable> TbCacheTransaction newTransactionForKey(String cacheName, K key) { |
|||
return null; |
|||
} |
|||
|
|||
@Override |
|||
public <K extends Serializable> TbCacheTransaction newTransactionForKeys(String cacheName, List<K> keys) { |
|||
return null; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,16 @@ |
|||
package org.thingsboard.server.cache; |
|||
|
|||
import com.google.common.util.concurrent.ListenableFuture; |
|||
|
|||
import java.util.concurrent.Executor; |
|||
|
|||
public interface TbCacheTransaction { |
|||
|
|||
<K,V> void putIfAbsent(K key, V value); |
|||
|
|||
boolean commit(); |
|||
|
|||
void rollback(); |
|||
|
|||
<T> void rollBackOnFailure(ListenableFuture<T> result, Executor cacheExecutor); |
|||
} |
|||
@ -0,0 +1,20 @@ |
|||
package org.thingsboard.server.cache; |
|||
|
|||
import org.springframework.cache.Cache; |
|||
|
|||
import java.io.Serializable; |
|||
import java.util.List; |
|||
|
|||
public interface TbTransactionalCache { |
|||
|
|||
<K extends Serializable> Cache.ValueWrapper get(String cacheName, K key); |
|||
|
|||
<K extends Serializable, V extends Serializable> void putIfAbsent(String cacheName, K key, V value); |
|||
|
|||
<K extends Serializable> void evict(String cacheName, K key); |
|||
|
|||
<K extends Serializable> TbCacheTransaction newTransactionForKey(String cacheName, K key); |
|||
|
|||
<K extends Serializable> TbCacheTransaction newTransactionForKeys(String cacheName, List<K> keys); |
|||
|
|||
} |
|||
@ -0,0 +1,36 @@ |
|||
/** |
|||
* 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.dao; |
|||
|
|||
import org.springframework.boot.autoconfigure.domain.EntityScan; |
|||
import org.springframework.context.annotation.ComponentScan; |
|||
import org.springframework.context.annotation.Configuration; |
|||
import org.springframework.data.jpa.repository.config.EnableJpaRepositories; |
|||
import org.springframework.transaction.annotation.EnableTransactionManagement; |
|||
import org.thingsboard.server.dao.util.TbAutoConfiguration; |
|||
|
|||
/** |
|||
* @author Valerii Sosliuk |
|||
*/ |
|||
@Configuration |
|||
@TbAutoConfiguration |
|||
@ComponentScan({"org.thingsboard.server.dao.sql", "org.thingsboard.server.dao.attributes"}) |
|||
@EnableJpaRepositories("org.thingsboard.server.dao.sql") |
|||
@EntityScan("org.thingsboard.server.dao.model.sql") |
|||
@EnableTransactionManagement |
|||
public class CacheDaoConfig { |
|||
|
|||
} |
|||
@ -0,0 +1,38 @@ |
|||
/** |
|||
* 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.dao; |
|||
|
|||
import org.springframework.boot.autoconfigure.domain.EntityScan; |
|||
import org.springframework.cache.annotation.EnableCaching; |
|||
import org.springframework.context.annotation.ComponentScan; |
|||
import org.springframework.context.annotation.Configuration; |
|||
import org.springframework.data.jpa.repository.config.EnableJpaRepositories; |
|||
import org.springframework.transaction.annotation.EnableTransactionManagement; |
|||
import org.thingsboard.server.dao.util.TbAutoConfiguration; |
|||
|
|||
/** |
|||
* @author Valerii Sosliuk |
|||
*/ |
|||
@Configuration |
|||
@EnableCaching |
|||
@TbAutoConfiguration |
|||
@ComponentScan({"org.thingsboard.server.dao.sql", "org.thingsboard.server.dao.attributes", "org.thingsboard.server.dao.cache", "org.thingsboard.server.cache"}) |
|||
@EnableJpaRepositories("org.thingsboard.server.dao.sql") |
|||
@EntityScan("org.thingsboard.server.dao.model.sql") |
|||
@EnableTransactionManagement |
|||
public class JpaServiceDaoConfig { |
|||
|
|||
} |
|||
@ -1,15 +0,0 @@ |
|||
package org.thingsboard.server.dao.attributes; |
|||
|
|||
import org.springframework.cache.Cache; |
|||
import org.thingsboard.server.common.data.kv.AttributeKvEntry; |
|||
|
|||
public interface AttributesCacheWrapper { |
|||
|
|||
Cache.ValueWrapper get(AttributeCacheKey attributeCacheKey); |
|||
|
|||
void put(AttributeCacheKey attributeCacheKey, AttributeKvEntry attributeKvEntry); |
|||
|
|||
void putIfAbsent(AttributeCacheKey attributeCacheKey, AttributeKvEntry attributeKvEntry); |
|||
|
|||
void evict(AttributeCacheKey attributeCacheKey); |
|||
} |
|||
@ -1,63 +0,0 @@ |
|||
/** |
|||
* 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.dao.attributes; |
|||
|
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; |
|||
import org.springframework.cache.Cache; |
|||
import org.springframework.cache.CacheManager; |
|||
import org.springframework.context.annotation.Primary; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.server.common.data.kv.AttributeKvEntry; |
|||
|
|||
import static org.thingsboard.server.common.data.CacheConstants.ATTRIBUTES_CACHE; |
|||
|
|||
@Service |
|||
@ConditionalOnProperty(prefix = "cache.attributes", value = "enabled", havingValue = "true") |
|||
@Primary |
|||
@Slf4j |
|||
public class DefaultAttributesCacheWrapper implements AttributesCacheWrapper { |
|||
private final Cache attributesCache; |
|||
|
|||
public DefaultAttributesCacheWrapper(CacheManager cacheManager) { |
|||
this.attributesCache = cacheManager.getCache(ATTRIBUTES_CACHE); |
|||
} |
|||
|
|||
@Override |
|||
public Cache.ValueWrapper get(AttributeCacheKey attributeCacheKey) { |
|||
var result = attributesCache.get(attributeCacheKey); |
|||
log.warn("[{}] Get = {}", attributeCacheKey, result); |
|||
return result; |
|||
} |
|||
|
|||
@Override |
|||
public void put(AttributeCacheKey attributeCacheKey, AttributeKvEntry attributeKvEntry) { |
|||
log.warn("[{}] Put = {}", attributeCacheKey, attributeKvEntry); |
|||
attributesCache.put(attributeCacheKey, attributeKvEntry); |
|||
} |
|||
|
|||
@Override |
|||
public void putIfAbsent(AttributeCacheKey attributeCacheKey, AttributeKvEntry attributeKvEntry) { |
|||
var result = attributesCache.putIfAbsent(attributeCacheKey, attributeKvEntry); |
|||
log.warn("[{}] Put if absent = {}, result = {}", attributeCacheKey, attributeKvEntry, result); |
|||
} |
|||
|
|||
@Override |
|||
public void evict(AttributeCacheKey attributeCacheKey) { |
|||
log.warn("[{}] Evict", attributeCacheKey); |
|||
attributesCache.evict(attributeCacheKey); |
|||
} |
|||
} |
|||
@ -0,0 +1,40 @@ |
|||
/** |
|||
* 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.dao; |
|||
|
|||
import org.junit.runner.RunWith; |
|||
import org.mockito.Answers; |
|||
import org.springframework.boot.test.mock.mockito.MockBean; |
|||
import org.springframework.test.context.ContextConfiguration; |
|||
import org.springframework.test.context.TestExecutionListeners; |
|||
import org.springframework.test.context.junit4.SpringRunner; |
|||
import org.springframework.test.context.support.DependencyInjectionTestExecutionListener; |
|||
import org.springframework.test.context.support.DirtiesContextTestExecutionListener; |
|||
import org.thingsboard.server.common.stats.StatsFactory; |
|||
import org.thingsboard.server.dao.service.DaoSqlTest; |
|||
|
|||
@RunWith(SpringRunner.class) |
|||
@ContextConfiguration(classes = {JpaServiceDaoConfig.class, PsqlTsDaoConfig.class, PsqlTsLatestDaoConfig.class, SqlTimeseriesDaoConfig.class}) |
|||
@DaoSqlTest |
|||
@TestExecutionListeners({ |
|||
DependencyInjectionTestExecutionListener.class, |
|||
DirtiesContextTestExecutionListener.class}) |
|||
public abstract class AbstractDaoServiceTest { |
|||
|
|||
@MockBean(answer = Answers.RETURNS_MOCKS) |
|||
StatsFactory statsFactory; |
|||
|
|||
} |
|||
@ -0,0 +1,50 @@ |
|||
/** |
|||
* 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.dao; |
|||
|
|||
import org.junit.ClassRule; |
|||
import org.junit.extensions.cpsuite.ClasspathSuite; |
|||
import org.junit.extensions.cpsuite.ClasspathSuite.ClassnameFilters; |
|||
import org.junit.runner.RunWith; |
|||
import org.springframework.context.ApplicationContextInitializer; |
|||
import org.springframework.context.ConfigurableApplicationContext; |
|||
import org.springframework.test.context.ContextConfiguration; |
|||
import org.springframework.test.context.support.TestPropertySourceUtils; |
|||
import org.testcontainers.containers.GenericContainer; |
|||
|
|||
@ContextConfiguration(initializers = RedisTestSuite.class) |
|||
@RunWith(ClasspathSuite.class) |
|||
@ClassnameFilters({ |
|||
"org.thingsboard.server.dao.sql.attributes.*ServiceTest", |
|||
}) |
|||
public class RedisTestSuite implements ApplicationContextInitializer<ConfigurableApplicationContext> { |
|||
|
|||
@ClassRule |
|||
public static GenericContainer redis = new GenericContainer("redis:4.0").withExposedPorts(6379); |
|||
|
|||
@Override |
|||
public void initialize(ConfigurableApplicationContext applicationContext) { |
|||
TestPropertySourceUtils.addInlinedPropertiesToEnvironment( |
|||
applicationContext, "cache.type=redis"); |
|||
TestPropertySourceUtils.addInlinedPropertiesToEnvironment( |
|||
applicationContext, "redis.connection.type=standalone"); |
|||
TestPropertySourceUtils.addInlinedPropertiesToEnvironment( |
|||
applicationContext, "redis.standalone.host=localhost"); |
|||
TestPropertySourceUtils.addInlinedPropertiesToEnvironment( |
|||
applicationContext, "redis.standalone.port=" + redis.getMappedPort(6379)); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,184 @@ |
|||
package org.thingsboard.server.dao.sql.attributes; |
|||
|
|||
import com.google.common.util.concurrent.Futures; |
|||
import com.google.common.util.concurrent.ListenableFuture; |
|||
import com.google.common.util.concurrent.ListeningExecutorService; |
|||
import com.google.common.util.concurrent.MoreExecutors; |
|||
import org.junit.Assert; |
|||
import org.junit.Test; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.thingsboard.server.common.data.DataConstants; |
|||
import org.thingsboard.server.common.data.id.DeviceId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.kv.AttributeKvEntry; |
|||
import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry; |
|||
import org.thingsboard.server.common.data.kv.StringDataEntry; |
|||
import org.thingsboard.server.dao.AbstractDaoServiceTest; |
|||
import org.thingsboard.server.dao.attributes.CachedAttributesService; |
|||
|
|||
import java.util.ArrayList; |
|||
import java.util.Arrays; |
|||
import java.util.Collections; |
|||
import java.util.List; |
|||
import java.util.Optional; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.Executors; |
|||
import java.util.concurrent.TimeUnit; |
|||
import java.util.stream.Collectors; |
|||
|
|||
public class AttributeServiceTest extends AbstractDaoServiceTest { |
|||
|
|||
private static final String OLD_VALUE = "OLD VALUE"; |
|||
private static final String NEW_VALUE = "NEW VALUE"; |
|||
|
|||
@Autowired |
|||
private CachedAttributesService attributesService; |
|||
|
|||
@Test |
|||
public void testDummyRequestWithEmptyResult() throws Exception { |
|||
var future = attributesService.find(new TenantId(UUID.randomUUID()), new DeviceId(UUID.randomUUID()), DataConstants.SERVER_SCOPE, "TEST"); |
|||
Assert.assertNotNull(future); |
|||
var result = future.get(10, TimeUnit.SECONDS); |
|||
Assert.assertTrue(result.isEmpty()); |
|||
} |
|||
|
|||
@Test |
|||
public void testConcurrentFetchAndUpdate() throws Exception { |
|||
var tenantId = new TenantId(UUID.randomUUID()); |
|||
ListeningExecutorService pool = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(2)); |
|||
try { |
|||
for (int i = 0; i < 100; i++) { |
|||
var deviceId = new DeviceId(UUID.randomUUID()); |
|||
testConcurrentFetchAndUpdate(tenantId, deviceId, pool); |
|||
} |
|||
} finally { |
|||
pool.shutdownNow(); |
|||
} |
|||
} |
|||
|
|||
@Test |
|||
public void testConcurrentFetchAndUpdateMulti() throws Exception { |
|||
var tenantId = new TenantId(UUID.randomUUID()); |
|||
ListeningExecutorService pool = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(2)); |
|||
try { |
|||
for (int i = 0; i < 100; i++) { |
|||
var deviceId = new DeviceId(UUID.randomUUID()); |
|||
testConcurrentFetchAndUpdateMulti(tenantId, deviceId, pool); |
|||
} |
|||
} finally { |
|||
pool.shutdownNow(); |
|||
} |
|||
} |
|||
|
|||
@Test |
|||
public void testFetchAndUpdateEmpty() throws Exception { |
|||
var tenantId = new TenantId(UUID.randomUUID()); |
|||
var deviceId = new DeviceId(UUID.randomUUID()); |
|||
var scope = DataConstants.SERVER_SCOPE; |
|||
var key = "TEST"; |
|||
|
|||
Optional<AttributeKvEntry> emptyValue = attributesService.find(tenantId, deviceId, scope, key).get(10, TimeUnit.SECONDS); |
|||
Assert.assertTrue(emptyValue.isEmpty()); |
|||
|
|||
saveAttribute(tenantId, deviceId, scope, key, NEW_VALUE); |
|||
Assert.assertEquals(NEW_VALUE, getAttributeValue(tenantId, deviceId, scope, key)); |
|||
} |
|||
|
|||
@Test |
|||
public void testFetchAndUpdateMulti() throws Exception { |
|||
var tenantId = new TenantId(UUID.randomUUID()); |
|||
var deviceId = new DeviceId(UUID.randomUUID()); |
|||
var scope = DataConstants.SERVER_SCOPE; |
|||
var key1 = "TEST1"; |
|||
var key2 = "TEST2"; |
|||
|
|||
var value = getAttributeValues(tenantId, deviceId, scope, Arrays.asList(key1, key2)); |
|||
Assert.assertTrue(value.isEmpty()); |
|||
|
|||
saveAttribute(tenantId, deviceId, scope, key1, OLD_VALUE); |
|||
|
|||
value = getAttributeValues(tenantId, deviceId, scope, Arrays.asList(key1, key2)); |
|||
Assert.assertEquals(1, value.size()); |
|||
Assert.assertEquals(OLD_VALUE, value.get(0)); |
|||
|
|||
saveAttribute(tenantId, deviceId, scope, key2, NEW_VALUE); |
|||
|
|||
value = getAttributeValues(tenantId, deviceId, scope, Arrays.asList(key1, key2)); |
|||
Assert.assertEquals(2, value.size()); |
|||
Assert.assertTrue(value.contains(OLD_VALUE)); |
|||
Assert.assertTrue(value.contains(NEW_VALUE)); |
|||
|
|||
saveAttribute(tenantId, deviceId, scope, key1, NEW_VALUE); |
|||
|
|||
value = getAttributeValues(tenantId, deviceId, scope, Arrays.asList(key1, key2)); |
|||
Assert.assertEquals(2, value.size()); |
|||
Assert.assertEquals(NEW_VALUE, value.get(0)); |
|||
Assert.assertEquals(NEW_VALUE, value.get(1)); |
|||
} |
|||
|
|||
private void testConcurrentFetchAndUpdate(TenantId tenantId, DeviceId deviceId, ListeningExecutorService pool) throws Exception { |
|||
var scope = DataConstants.SERVER_SCOPE; |
|||
var key = "TEST"; |
|||
saveAttribute(tenantId, deviceId, scope, key, OLD_VALUE); |
|||
List<ListenableFuture<?>> futures = new ArrayList<>(); |
|||
futures.add(pool.submit(() -> { |
|||
var value = getAttributeValue(tenantId, deviceId, scope, key); |
|||
Assert.assertTrue(value.equals(OLD_VALUE) || value.equals(NEW_VALUE)); |
|||
})); |
|||
futures.add(pool.submit(() -> saveAttribute(tenantId, deviceId, scope, key, NEW_VALUE))); |
|||
Futures.allAsList(futures).get(10, TimeUnit.SECONDS); |
|||
Assert.assertEquals(NEW_VALUE, getAttributeValue(tenantId, deviceId, scope, key)); |
|||
} |
|||
|
|||
private void testConcurrentFetchAndUpdateMulti(TenantId tenantId, DeviceId deviceId, ListeningExecutorService pool) throws Exception { |
|||
var scope = DataConstants.SERVER_SCOPE; |
|||
var key1 = "TEST1"; |
|||
var key2 = "TEST2"; |
|||
saveAttribute(tenantId, deviceId, scope, key1, OLD_VALUE); |
|||
saveAttribute(tenantId, deviceId, scope, key2, OLD_VALUE); |
|||
List<ListenableFuture<?>> futures = new ArrayList<>(); |
|||
futures.add(pool.submit(() -> { |
|||
var value = getAttributeValues(tenantId, deviceId, scope, Arrays.asList(key1, key2)); |
|||
Assert.assertEquals(2, value.size()); |
|||
Assert.assertTrue(value.contains(OLD_VALUE) || value.contains(NEW_VALUE)); |
|||
})); |
|||
futures.add(pool.submit(() -> { |
|||
saveAttribute(tenantId, deviceId, scope, key1, NEW_VALUE); |
|||
saveAttribute(tenantId, deviceId, scope, key2, NEW_VALUE); |
|||
})); |
|||
Futures.allAsList(futures).get(10, TimeUnit.SECONDS); |
|||
var newResult = getAttributeValues(tenantId, deviceId, scope, Arrays.asList(key1, key2)); |
|||
Assert.assertEquals(2, newResult.size()); |
|||
Assert.assertEquals(NEW_VALUE, newResult.get(0)); |
|||
Assert.assertEquals(NEW_VALUE, newResult.get(1)); |
|||
} |
|||
|
|||
private String getAttributeValue(TenantId tenantId, DeviceId deviceId, String scope, String key) { |
|||
try { |
|||
Optional<AttributeKvEntry> entry = attributesService.find(tenantId, deviceId, scope, key).get(10, TimeUnit.SECONDS); |
|||
return entry.orElseThrow(RuntimeException::new).getStrValue().orElse("Unknown"); |
|||
} catch (Exception e) { |
|||
throw new RuntimeException(e); |
|||
} |
|||
} |
|||
|
|||
private List<String> getAttributeValues(TenantId tenantId, DeviceId deviceId, String scope, List<String> keys) { |
|||
try { |
|||
List<AttributeKvEntry> entry = attributesService.find(tenantId, deviceId, scope, keys).get(10, TimeUnit.SECONDS); |
|||
return entry.stream().map(e -> e.getStrValue().orElse(null)).collect(Collectors.toList()); |
|||
} catch (Exception e) { |
|||
throw new RuntimeException(e); |
|||
} |
|||
} |
|||
|
|||
private void saveAttribute(TenantId tenantId, DeviceId deviceId, String scope, String key, String s) { |
|||
try { |
|||
AttributeKvEntry newEntry = new BaseAttributeKvEntry(System.currentTimeMillis(), new StringDataEntry(key, s)); |
|||
attributesService.save(tenantId, deviceId, scope, Collections.singletonList(newEntry)).get(10, TimeUnit.SECONDS); |
|||
} catch (Exception e) { |
|||
Assert.assertNull(e); |
|||
} |
|||
} |
|||
|
|||
|
|||
} |
|||
@ -0,0 +1,30 @@ |
|||
package org.thingsboard.server.dao.sql.attributes; |
|||
|
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.junit.ClassRule; |
|||
import org.springframework.context.ApplicationContextInitializer; |
|||
import org.springframework.context.ConfigurableApplicationContext; |
|||
import org.springframework.test.context.ContextConfiguration; |
|||
import org.springframework.test.context.TestPropertySource; |
|||
import org.springframework.test.context.support.TestPropertySourceUtils; |
|||
import org.testcontainers.containers.GenericContainer; |
|||
|
|||
@TestPropertySource(properties = { |
|||
"cache.type=redis", "redis.connection.type=standalone" |
|||
}) |
|||
@ContextConfiguration(initializers = RedisAttributeServiceTest.class) |
|||
@Slf4j |
|||
public class RedisAttributeServiceTest extends AttributeServiceTest implements ApplicationContextInitializer<ConfigurableApplicationContext> { |
|||
|
|||
@ClassRule |
|||
public static GenericContainer redis = new GenericContainer("redis:4.0").withExposedPorts(6379); |
|||
|
|||
@Override |
|||
public void initialize(ConfigurableApplicationContext applicationContext) { |
|||
TestPropertySourceUtils.addInlinedPropertiesToEnvironment( |
|||
applicationContext, "redis.standalone.host=localhost"); |
|||
TestPropertySourceUtils.addInlinedPropertiesToEnvironment( |
|||
applicationContext, "redis.standalone.port=" + redis.getMappedPort(6379)); |
|||
} |
|||
|
|||
} |
|||
Loading…
Reference in new issue