Andrii Shvaika
2 years ago
25 changed files with 578 additions and 257 deletions
@ -0,0 +1,35 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.cf.cache; |
|||
|
|||
import org.springframework.context.ApplicationListener; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.queue.discovery.event.PartitionChangeEvent; |
|||
|
|||
import java.util.Collection; |
|||
|
|||
public interface CalculatedFieldEntityProfileCache extends ApplicationListener<PartitionChangeEvent> { |
|||
|
|||
void add(TenantId tenantId, EntityId profileId, EntityId entityId); |
|||
|
|||
void update(TenantId tenantId, EntityId oldProfileId, EntityId newProfileId, EntityId entityId); |
|||
|
|||
void evict(TenantId tenantId, EntityId entityId); |
|||
|
|||
Collection<EntityId> getMyEntityIdsByProfileId(TenantId tenantId, EntityId profileId); |
|||
|
|||
} |
|||
@ -0,0 +1,86 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.cf.cache; |
|||
|
|||
import lombok.RequiredArgsConstructor; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; |
|||
import org.thingsboard.server.queue.discovery.HashPartitionService; |
|||
import org.thingsboard.server.queue.discovery.PartitionService; |
|||
import org.thingsboard.server.queue.discovery.TbApplicationEventListener; |
|||
import org.thingsboard.server.queue.discovery.event.PartitionChangeEvent; |
|||
import org.thingsboard.server.queue.util.TbRuleEngineComponent; |
|||
|
|||
import java.util.Collection; |
|||
import java.util.Collections; |
|||
import java.util.List; |
|||
import java.util.Set; |
|||
import java.util.concurrent.ConcurrentHashMap; |
|||
import java.util.concurrent.ConcurrentMap; |
|||
import java.util.stream.Collectors; |
|||
|
|||
@TbRuleEngineComponent |
|||
@Service |
|||
@Slf4j |
|||
@RequiredArgsConstructor |
|||
public class DefaultCalculatedFieldEntityProfileCache extends TbApplicationEventListener<PartitionChangeEvent> implements CalculatedFieldEntityProfileCache { |
|||
|
|||
private static final Integer UNKNOWN = -1; |
|||
private final ConcurrentMap<TenantId, TenantEntityProfileCache> tenantCache = new ConcurrentHashMap<>(); |
|||
private final PartitionService partitionService; |
|||
private volatile List<Integer> myPartitions = Collections.emptyList(); |
|||
|
|||
@Override |
|||
protected void onTbApplicationEvent(PartitionChangeEvent event) { |
|||
myPartitions = event.getCalculatedFieldsPartitions().stream() |
|||
.filter(TopicPartitionInfo::isMyPartition) |
|||
.map(tpi -> tpi.getPartition().orElse(UNKNOWN)).collect(Collectors.toList()); |
|||
//Naive approach that need to be improved.
|
|||
tenantCache.values().forEach(cache -> cache.setMyPartitions(myPartitions)); |
|||
} |
|||
|
|||
@Override |
|||
public void add(TenantId tenantId, EntityId profileId, EntityId entityId) { |
|||
var tpi = partitionService.resolve(HashPartitionService.CALCULATED_FIELD_QUEUE_KEY, entityId); |
|||
var partition = tpi.getPartition().orElse(UNKNOWN); |
|||
tenantCache.computeIfAbsent(tenantId, id -> new TenantEntityProfileCache()) |
|||
.add(profileId, entityId, partition, tpi.isMyPartition()); |
|||
} |
|||
|
|||
@Override |
|||
public void update(TenantId tenantId, EntityId oldProfileId, EntityId newProfileId, EntityId entityId) { |
|||
var tpi = partitionService.resolve(HashPartitionService.CALCULATED_FIELD_QUEUE_KEY, entityId); |
|||
var partition = tpi.getPartition().orElse(UNKNOWN); |
|||
var cache = tenantCache.computeIfAbsent(tenantId, id -> new TenantEntityProfileCache()); |
|||
//TODO: make this method atomic;
|
|||
cache.remove(oldProfileId, entityId); |
|||
cache.add(newProfileId, entityId, partition, tpi.isMyPartition()); |
|||
} |
|||
|
|||
@Override |
|||
public void evict(TenantId tenantId, EntityId entityId) { |
|||
var cache = tenantCache.computeIfAbsent(tenantId, id -> new TenantEntityProfileCache()); |
|||
cache.removeEntityId(entityId); |
|||
} |
|||
|
|||
@Override |
|||
public Collection<EntityId> getMyEntityIdsByProfileId(TenantId tenantId, EntityId profileId) { |
|||
return tenantCache.computeIfAbsent(tenantId, id -> new TenantEntityProfileCache()).getMyEntityIdsByProfileId(profileId); |
|||
} |
|||
} |
|||
@ -0,0 +1,118 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.cf.cache; |
|||
|
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
|
|||
import java.util.ArrayList; |
|||
import java.util.Collection; |
|||
import java.util.Collections; |
|||
import java.util.HashMap; |
|||
import java.util.HashSet; |
|||
import java.util.List; |
|||
import java.util.Map; |
|||
import java.util.Set; |
|||
import java.util.concurrent.locks.ReadWriteLock; |
|||
import java.util.concurrent.locks.ReentrantReadWriteLock; |
|||
|
|||
public class TenantEntityProfileCache { |
|||
|
|||
private final ReadWriteLock lock = new ReentrantReadWriteLock(); |
|||
private final Map<Integer, Map<EntityId, Set<EntityId>>> allEntities = new HashMap<>(); |
|||
private final Map<EntityId, Set<EntityId>> myEntities = new HashMap<>(); |
|||
|
|||
public void setMyPartitions(List<Integer> myPartitions) { |
|||
lock.writeLock().lock(); |
|||
try { |
|||
myEntities.clear(); |
|||
myPartitions.forEach(partitionId -> { |
|||
var map = allEntities.get(partitionId); |
|||
if (map != null) { |
|||
map.forEach((profileId, entityIds) -> myEntities.computeIfAbsent(profileId, k -> new HashSet<>()).addAll(entityIds)); |
|||
} |
|||
}); |
|||
} finally { |
|||
lock.writeLock().unlock(); |
|||
} |
|||
} |
|||
|
|||
public void removeProfileId(EntityId profileId) { |
|||
lock.writeLock().lock(); |
|||
try { |
|||
// Remove from allEntities
|
|||
allEntities.values().forEach(map -> map.remove(profileId)); |
|||
// Remove from myEntities
|
|||
myEntities.remove(profileId); |
|||
} finally { |
|||
lock.writeLock().unlock(); |
|||
} |
|||
} |
|||
|
|||
public void removeEntityId(EntityId entityId) { |
|||
lock.writeLock().lock(); |
|||
try { |
|||
// Remove from allEntities
|
|||
allEntities.values().forEach(map -> map.values().forEach(set -> set.remove(entityId))); |
|||
// Remove from myEntities
|
|||
myEntities.values().forEach(set -> set.remove(entityId)); |
|||
} finally { |
|||
lock.writeLock().unlock(); |
|||
} |
|||
} |
|||
|
|||
public void remove(EntityId profileId, EntityId entityId) { |
|||
lock.writeLock().lock(); |
|||
try { |
|||
// Remove from allEntities
|
|||
allEntities.values().forEach(map -> removeSafely(map, profileId, entityId)); |
|||
// Remove from myEntities
|
|||
removeSafely(myEntities, profileId, entityId); |
|||
} finally { |
|||
lock.writeLock().unlock(); |
|||
} |
|||
} |
|||
|
|||
public void add(EntityId profileId, EntityId entityId, Integer partition, boolean mine) { |
|||
lock.writeLock().lock(); |
|||
try { |
|||
if (mine) { |
|||
myEntities.computeIfAbsent(profileId, k -> new HashSet<>()).add(entityId); |
|||
} |
|||
allEntities.computeIfAbsent(partition, k -> new HashMap<>()).computeIfAbsent(profileId, p -> new HashSet<>()).add(entityId); |
|||
} finally { |
|||
lock.writeLock().unlock(); |
|||
} |
|||
} |
|||
|
|||
public Collection<EntityId> getMyEntityIdsByProfileId(EntityId profileId) { |
|||
lock.readLock().lock(); |
|||
try { |
|||
var entities = myEntities.getOrDefault(profileId, Collections.emptySet()); |
|||
List<EntityId> result = new ArrayList<>(entities.size()); |
|||
result.addAll(entities); |
|||
return result; |
|||
} finally { |
|||
lock.readLock().unlock(); |
|||
} |
|||
} |
|||
|
|||
private void removeSafely(Map<EntityId, Set<EntityId>> map, EntityId profileId, EntityId entityId) { |
|||
var set = map.get(profileId); |
|||
if (set != null) { |
|||
set.remove(entityId); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,55 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.common.data; |
|||
|
|||
import lombok.Data; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.thingsboard.server.common.data.id.AssetId; |
|||
import org.thingsboard.server.common.data.id.AssetProfileId; |
|||
import org.thingsboard.server.common.data.id.CustomerId; |
|||
import org.thingsboard.server.common.data.id.DeviceId; |
|||
import org.thingsboard.server.common.data.id.DeviceProfileId; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
|
|||
import java.io.Serializable; |
|||
import java.util.UUID; |
|||
|
|||
@Data |
|||
@Slf4j |
|||
public class ProfileEntityIdInfo implements Serializable, HasTenantId { |
|||
|
|||
private static final long serialVersionUID = 8532058281983868003L; |
|||
|
|||
private final TenantId tenantId; |
|||
private final EntityId profileId; |
|||
private final EntityId entityId; |
|||
|
|||
private ProfileEntityIdInfo(UUID tenantId, EntityId profileId, EntityId entityId) { |
|||
this.tenantId = TenantId.fromUUID(tenantId); |
|||
this.profileId = profileId; |
|||
this.entityId = entityId; |
|||
} |
|||
|
|||
public static ProfileEntityIdInfo create(UUID tenantId, DeviceProfileId profileId, DeviceId entityId) { |
|||
return new ProfileEntityIdInfo(tenantId, profileId, entityId); |
|||
} |
|||
|
|||
public static ProfileEntityIdInfo create(UUID tenantId, AssetProfileId profileId, AssetId entityId) { |
|||
return new ProfileEntityIdInfo(tenantId, profileId, entityId); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,54 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.sql.device; |
|||
|
|||
import lombok.RequiredArgsConstructor; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.data.domain.Pageable; |
|||
import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate; |
|||
import org.springframework.stereotype.Repository; |
|||
import org.springframework.transaction.support.TransactionTemplate; |
|||
import org.thingsboard.server.common.data.page.PageData; |
|||
|
|||
import java.util.Collections; |
|||
import java.util.List; |
|||
import java.util.Map; |
|||
import java.util.function.Function; |
|||
import java.util.stream.Collectors; |
|||
|
|||
@RequiredArgsConstructor |
|||
@Repository |
|||
@Slf4j |
|||
public class AbstractNativeRepository { |
|||
|
|||
private final NamedParameterJdbcTemplate jdbcTemplate; |
|||
private final TransactionTemplate transactionTemplate; |
|||
|
|||
protected <T> PageData<T> find(String countQuery, String findQuery, Pageable pageable, Function<Map<String, Object>, T> mapper) { |
|||
return transactionTemplate.execute(status -> { |
|||
long startTs = System.currentTimeMillis(); |
|||
int totalElements = jdbcTemplate.queryForObject(countQuery, Collections.emptyMap(), Integer.class); |
|||
log.debug("Count query took {} ms", System.currentTimeMillis() - startTs); |
|||
startTs = System.currentTimeMillis(); |
|||
List<Map<String, Object>> rows = jdbcTemplate.queryForList(String.format(findQuery, pageable.getPageSize(), pageable.getOffset()), Collections.emptyMap()); |
|||
log.debug("Main query took {} ms", System.currentTimeMillis() - startTs); |
|||
int totalPages = pageable.getPageSize() > 0 ? (int) Math.ceil((float) totalElements / pageable.getPageSize()) : 1; |
|||
boolean hasNext = pageable.getPageSize() > 0 && totalElements > pageable.getOffset() + rows.size(); |
|||
var data = rows.stream().map(mapper).collect(Collectors.toList()); |
|||
return new PageData<>(data, totalPages, totalElements, hasNext); |
|||
}); |
|||
} |
|||
} |
|||
@ -0,0 +1,55 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.sql.device; |
|||
|
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.data.domain.Pageable; |
|||
import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate; |
|||
import org.springframework.stereotype.Repository; |
|||
import org.springframework.transaction.support.TransactionTemplate; |
|||
import org.thingsboard.server.common.data.DeviceIdInfo; |
|||
import org.thingsboard.server.common.data.ProfileEntityIdInfo; |
|||
import org.thingsboard.server.common.data.id.AssetId; |
|||
import org.thingsboard.server.common.data.id.AssetProfileId; |
|||
import org.thingsboard.server.common.data.id.DeviceId; |
|||
import org.thingsboard.server.common.data.id.DeviceProfileId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.page.PageData; |
|||
|
|||
import java.util.UUID; |
|||
|
|||
@Repository |
|||
@Slf4j |
|||
public class DefaultNativeAssetRepository extends AbstractNativeRepository implements NativeAssetRepository { |
|||
|
|||
private final String COUNT_QUERY = "SELECT count(id) FROM asset;"; |
|||
|
|||
public DefaultNativeAssetRepository(NamedParameterJdbcTemplate jdbcTemplate, TransactionTemplate transactionTemplate) { |
|||
super(jdbcTemplate, transactionTemplate); |
|||
} |
|||
|
|||
@Override |
|||
public PageData<ProfileEntityIdInfo> findProfileEntityIdInfos(Pageable pageable) { |
|||
String PROFILE_DEVICE_ID_INFO_QUERY = "SELECT tenant_id as tenantId, asset_profile_id as profileId, id as id FROM asset ORDER BY created_time ASC LIMIT %s OFFSET %s"; |
|||
return find(COUNT_QUERY, PROFILE_DEVICE_ID_INFO_QUERY, pageable, row -> { |
|||
AssetId id = new AssetId((UUID) row.get("id")); |
|||
AssetProfileId profileId = new AssetProfileId((UUID) row.get("profileId")); |
|||
var tenantIdObj = row.get("tenantId"); |
|||
return ProfileEntityIdInfo.create(tenantIdObj != null ? (UUID) tenantIdObj : TenantId.SYS_TENANT_ID.getId(), profileId, id); |
|||
}); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,20 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.sql.device; |
|||
|
|||
public interface NativeAssetRepository extends NativeProfileEntityRepository { |
|||
|
|||
} |
|||
@ -0,0 +1,26 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.sql.device; |
|||
|
|||
import org.springframework.data.domain.Pageable; |
|||
import org.thingsboard.server.common.data.ProfileEntityIdInfo; |
|||
import org.thingsboard.server.common.data.page.PageData; |
|||
|
|||
public interface NativeProfileEntityRepository { |
|||
|
|||
PageData<ProfileEntityIdInfo> findProfileEntityIdInfos(Pageable pageable); |
|||
|
|||
} |
|||
Loading…
Reference in new issue