495 changed files with 18291 additions and 1653 deletions
@ -0,0 +1,82 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2025 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.edge.rpc.processor.rule; |
||||
|
|
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.springframework.beans.factory.annotation.Autowired; |
||||
|
import org.springframework.data.util.Pair; |
||||
|
import org.thingsboard.common.util.JacksonUtil; |
||||
|
import org.thingsboard.server.common.data.id.RuleChainId; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
import org.thingsboard.server.common.data.rule.RuleChain; |
||||
|
import org.thingsboard.server.common.data.rule.RuleChainMetaData; |
||||
|
import org.thingsboard.server.common.data.rule.RuleChainType; |
||||
|
import org.thingsboard.server.common.data.rule.RuleNode; |
||||
|
import org.thingsboard.server.dao.service.DataValidator; |
||||
|
import org.thingsboard.server.gen.edge.v1.RuleChainMetadataUpdateMsg; |
||||
|
import org.thingsboard.server.gen.edge.v1.RuleChainUpdateMsg; |
||||
|
import org.thingsboard.server.service.edge.rpc.processor.BaseEdgeProcessor; |
||||
|
|
||||
|
import java.util.function.Function; |
||||
|
|
||||
|
@Slf4j |
||||
|
public class BaseRuleChainProcessor extends BaseEdgeProcessor { |
||||
|
|
||||
|
@Autowired |
||||
|
private DataValidator<RuleChain> ruleChainValidator; |
||||
|
|
||||
|
protected Pair<Boolean, Boolean> saveOrUpdateRuleChain(TenantId tenantId, RuleChainId ruleChainId, RuleChainUpdateMsg ruleChainUpdateMsg, RuleChainType ruleChainType) { |
||||
|
boolean created = false; |
||||
|
RuleChain ruleChainFromDb = edgeCtx.getRuleChainService().findRuleChainById(tenantId, ruleChainId); |
||||
|
if (ruleChainFromDb == null) { |
||||
|
created = true; |
||||
|
} |
||||
|
|
||||
|
RuleChain ruleChain = JacksonUtil.fromString(ruleChainUpdateMsg.getEntity(), RuleChain.class, true); |
||||
|
if (ruleChain == null) { |
||||
|
throw new RuntimeException("[{" + tenantId + "}] ruleChainUpdateMsg {" + ruleChainUpdateMsg + "} cannot be converted to rule chain"); |
||||
|
} |
||||
|
boolean isRoot = ruleChain.isRoot(); |
||||
|
if (RuleChainType.CORE.equals(ruleChainType)) { |
||||
|
ruleChain.setRoot(false); |
||||
|
} else { |
||||
|
ruleChain.setRoot(ruleChainFromDb == null ? false : ruleChainFromDb.isRoot()); |
||||
|
} |
||||
|
ruleChain.setType(ruleChainType); |
||||
|
|
||||
|
ruleChainValidator.validate(ruleChain, RuleChain::getTenantId); |
||||
|
if (created) { |
||||
|
ruleChain.setId(ruleChainId); |
||||
|
} |
||||
|
edgeCtx.getRuleChainService().saveRuleChain(ruleChain, true, false); |
||||
|
return Pair.of(created, isRoot); |
||||
|
} |
||||
|
|
||||
|
protected void saveOrUpdateRuleChainMetadata(TenantId tenantId, RuleChainMetadataUpdateMsg ruleChainMetadataUpdateMsg) { |
||||
|
RuleChainMetaData ruleChainMetadata = JacksonUtil.fromString(ruleChainMetadataUpdateMsg.getEntity(), RuleChainMetaData.class, true); |
||||
|
if (ruleChainMetadata == null) { |
||||
|
throw new RuntimeException("[{" + tenantId + "}] ruleChainMetadataUpdateMsg {" + ruleChainMetadataUpdateMsg + "} cannot be converted to rule chain metadata"); |
||||
|
} |
||||
|
if (!ruleChainMetadata.getNodes().isEmpty()) { |
||||
|
ruleChainMetadata.setVersion(null); |
||||
|
for (RuleNode ruleNode : ruleChainMetadata.getNodes()) { |
||||
|
ruleNode.setRuleChainId(null); |
||||
|
ruleNode.setId(null); |
||||
|
} |
||||
|
edgeCtx.getRuleChainService().saveRuleChainMetaData(tenantId, ruleChainMetadata, Function.identity(), true); |
||||
|
} |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,117 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2025 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.edqs; |
||||
|
|
||||
|
import com.google.common.util.concurrent.Futures; |
||||
|
import com.google.common.util.concurrent.ListenableFuture; |
||||
|
import com.google.common.util.concurrent.MoreExecutors; |
||||
|
import jakarta.annotation.PostConstruct; |
||||
|
import jakarta.annotation.PreDestroy; |
||||
|
import lombok.RequiredArgsConstructor; |
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.springframework.beans.factory.annotation.Value; |
||||
|
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; |
||||
|
import org.springframework.stereotype.Service; |
||||
|
import org.thingsboard.common.util.JacksonUtil; |
||||
|
import org.thingsboard.server.common.data.edqs.query.EdqsRequest; |
||||
|
import org.thingsboard.server.common.data.edqs.query.EdqsResponse; |
||||
|
import org.thingsboard.server.common.data.id.CustomerId; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
import org.thingsboard.server.common.msg.edqs.EdqsApiService; |
||||
|
import org.thingsboard.server.edqs.state.EdqsPartitionService; |
||||
|
import org.thingsboard.server.gen.transport.TransportProtos; |
||||
|
import org.thingsboard.server.gen.transport.TransportProtos.FromEdqsMsg; |
||||
|
import org.thingsboard.server.gen.transport.TransportProtos.ToEdqsMsg; |
||||
|
import org.thingsboard.server.queue.TbQueueRequestTemplate; |
||||
|
import org.thingsboard.server.queue.common.TbProtoQueueMsg; |
||||
|
import org.thingsboard.server.queue.provider.EdqsClientQueueFactory; |
||||
|
|
||||
|
import java.util.UUID; |
||||
|
|
||||
|
@Service |
||||
|
@Slf4j |
||||
|
@RequiredArgsConstructor |
||||
|
@ConditionalOnExpression("'${queue.edqs.api.supported:true}' == 'true' && ('${service.type:null}' == 'monolith' || '${service.type:null}' == 'tb-core')") |
||||
|
public class DefaultEdqsApiService implements EdqsApiService { |
||||
|
|
||||
|
private final EdqsPartitionService edqsPartitionService; |
||||
|
private final EdqsClientQueueFactory queueFactory; |
||||
|
private TbQueueRequestTemplate<TbProtoQueueMsg<ToEdqsMsg>, TbProtoQueueMsg<FromEdqsMsg>> requestTemplate; |
||||
|
|
||||
|
@Value("${queue.edqs.api.auto_enable:true}") |
||||
|
private boolean autoEnable; |
||||
|
|
||||
|
private Boolean apiEnabled = null; |
||||
|
|
||||
|
@PostConstruct |
||||
|
private void init() { |
||||
|
requestTemplate = queueFactory.createEdqsRequestTemplate(); |
||||
|
requestTemplate.init(); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public ListenableFuture<EdqsResponse> processRequest(TenantId tenantId, CustomerId customerId, EdqsRequest request) { |
||||
|
var requestMsg = ToEdqsMsg.newBuilder() |
||||
|
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
||||
|
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
||||
|
.setTs(System.currentTimeMillis()) |
||||
|
.setRequestMsg(TransportProtos.EdqsRequestMsg.newBuilder() |
||||
|
.setValue(JacksonUtil.toString(request)) |
||||
|
.build()); |
||||
|
if (customerId != null && !customerId.isNullUid()) { |
||||
|
requestMsg.setCustomerIdMSB(customerId.getId().getMostSignificantBits()); |
||||
|
requestMsg.setCustomerIdLSB(customerId.getId().getLeastSignificantBits()); |
||||
|
} |
||||
|
|
||||
|
Integer partition = edqsPartitionService.resolvePartition(tenantId); |
||||
|
ListenableFuture<TbProtoQueueMsg<FromEdqsMsg>> resultFuture = requestTemplate.send(new TbProtoQueueMsg<>(UUID.randomUUID(), requestMsg.build()), partition); |
||||
|
return Futures.transform(resultFuture, msg -> { |
||||
|
TransportProtos.EdqsResponseMsg responseMsg = msg.getValue().getResponseMsg(); |
||||
|
return JacksonUtil.fromString(responseMsg.getValue(), EdqsResponse.class); |
||||
|
}, MoreExecutors.directExecutor()); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public boolean isEnabled() { |
||||
|
return Boolean.TRUE.equals(apiEnabled); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public void setEnabled(boolean enabled) { |
||||
|
if (enabled) { |
||||
|
log.info("Enabling EDQS API"); |
||||
|
} else { |
||||
|
log.info("Disabling EDQS API"); |
||||
|
} |
||||
|
apiEnabled = enabled; |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public boolean isSupported() { |
||||
|
return true; |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public boolean isAutoEnable() { |
||||
|
return autoEnable; |
||||
|
} |
||||
|
|
||||
|
@PreDestroy |
||||
|
private void stop() { |
||||
|
requestTemplate.stop(); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,298 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2025 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.edqs; |
||||
|
|
||||
|
import com.google.protobuf.ByteString; |
||||
|
import jakarta.annotation.PostConstruct; |
||||
|
import jakarta.annotation.PreDestroy; |
||||
|
import lombok.AllArgsConstructor; |
||||
|
import lombok.Data; |
||||
|
import lombok.NoArgsConstructor; |
||||
|
import lombok.RequiredArgsConstructor; |
||||
|
import lombok.SneakyThrows; |
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.springframework.beans.factory.annotation.Autowired; |
||||
|
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; |
||||
|
import org.springframework.context.annotation.Lazy; |
||||
|
import org.springframework.stereotype.Service; |
||||
|
import org.thingsboard.common.util.JacksonUtil; |
||||
|
import org.thingsboard.common.util.ThingsBoardExecutors; |
||||
|
import org.thingsboard.server.cluster.TbClusterService; |
||||
|
import org.thingsboard.server.common.data.AttributeScope; |
||||
|
import org.thingsboard.server.common.data.EntityType; |
||||
|
import org.thingsboard.server.common.data.ObjectType; |
||||
|
import org.thingsboard.server.common.data.edqs.EdqsEventType; |
||||
|
import org.thingsboard.server.common.data.edqs.EdqsObject; |
||||
|
import org.thingsboard.server.common.data.edqs.EdqsSyncRequest; |
||||
|
import org.thingsboard.server.common.data.edqs.Entity; |
||||
|
import org.thingsboard.server.common.data.edqs.ToCoreEdqsMsg; |
||||
|
import org.thingsboard.server.common.data.edqs.ToCoreEdqsRequest; |
||||
|
import org.thingsboard.server.common.data.id.EntityId; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry; |
||||
|
import org.thingsboard.server.common.data.kv.JsonDataEntry; |
||||
|
import org.thingsboard.server.common.data.kv.KvEntry; |
||||
|
import org.thingsboard.server.common.msg.edqs.EdqsApiService; |
||||
|
import org.thingsboard.server.common.msg.edqs.EdqsService; |
||||
|
import org.thingsboard.server.common.msg.queue.ServiceType; |
||||
|
import org.thingsboard.server.dao.attributes.AttributesService; |
||||
|
import org.thingsboard.server.edqs.processor.EdqsProducer; |
||||
|
import org.thingsboard.server.edqs.state.EdqsPartitionService; |
||||
|
import org.thingsboard.server.edqs.util.EdqsConverter; |
||||
|
import org.thingsboard.server.gen.transport.TransportProtos.EdqsEventMsg; |
||||
|
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; |
||||
|
import org.thingsboard.server.gen.transport.TransportProtos.ToEdqsCoreServiceMsg; |
||||
|
import org.thingsboard.server.gen.transport.TransportProtos.ToEdqsMsg; |
||||
|
import org.thingsboard.server.queue.discovery.HashPartitionService; |
||||
|
import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; |
||||
|
import org.thingsboard.server.queue.discovery.TopicService; |
||||
|
import org.thingsboard.server.queue.edqs.EdqsQueue; |
||||
|
import org.thingsboard.server.queue.environment.DistributedLock; |
||||
|
import org.thingsboard.server.queue.environment.DistributedLockService; |
||||
|
import org.thingsboard.server.queue.provider.EdqsClientQueueFactory; |
||||
|
import org.thingsboard.server.queue.util.AfterStartUp; |
||||
|
|
||||
|
import java.util.concurrent.ExecutorService; |
||||
|
import java.util.concurrent.TimeUnit; |
||||
|
|
||||
|
@Service |
||||
|
@RequiredArgsConstructor |
||||
|
@Slf4j |
||||
|
@ConditionalOnProperty(value = "queue.edqs.sync.enabled", havingValue = "true") |
||||
|
public class DefaultEdqsService implements EdqsService { |
||||
|
|
||||
|
private final EdqsClientQueueFactory queueFactory; |
||||
|
private final EdqsConverter edqsConverter; |
||||
|
private final EdqsSyncService edqsSyncService; |
||||
|
private final EdqsApiService edqsApiService; |
||||
|
private final DistributedLockService distributedLockService; |
||||
|
private final AttributesService attributesService; |
||||
|
private final EdqsPartitionService edqsPartitionService; |
||||
|
private final TopicService topicService; |
||||
|
private final TbServiceInfoProvider serviceInfoProvider; |
||||
|
@Autowired @Lazy |
||||
|
private TbClusterService clusterService; |
||||
|
@Autowired @Lazy |
||||
|
private HashPartitionService hashPartitionService; |
||||
|
|
||||
|
private EdqsProducer eventsProducer; |
||||
|
private ExecutorService executor; |
||||
|
private DistributedLock syncLock; |
||||
|
|
||||
|
@PostConstruct |
||||
|
private void init() { |
||||
|
executor = ThingsBoardExecutors.newWorkStealingPool(12, getClass()); |
||||
|
eventsProducer = EdqsProducer.builder() |
||||
|
.queue(EdqsQueue.EVENTS) |
||||
|
.partitionService(edqsPartitionService) |
||||
|
.topicService(topicService) |
||||
|
.producer(queueFactory.createEdqsMsgProducer(EdqsQueue.EVENTS)) |
||||
|
.build(); |
||||
|
syncLock = distributedLockService.getLock("edqs_sync"); |
||||
|
} |
||||
|
|
||||
|
@AfterStartUp(order = AfterStartUp.REGULAR_SERVICE) |
||||
|
public void onStartUp() { |
||||
|
if (!serviceInfoProvider.isService(ServiceType.TB_CORE)) { |
||||
|
return; |
||||
|
} |
||||
|
executor.submit(() -> { |
||||
|
try { |
||||
|
EdqsSyncState syncState = getSyncState(); |
||||
|
if (edqsSyncService.isSyncNeeded() || syncState == null || syncState.getStatus() != EdqsSyncStatus.FINISHED) { |
||||
|
if (hashPartitionService.isSystemPartitionMine(ServiceType.TB_CORE)) { |
||||
|
processSystemRequest(ToCoreEdqsRequest.builder() |
||||
|
.syncRequest(new EdqsSyncRequest()) |
||||
|
.build()); |
||||
|
} |
||||
|
} else if (edqsApiService.isSupported() && edqsApiService.isAutoEnable()) { |
||||
|
// only if topic/RocksDB is not empty and sync is finished
|
||||
|
edqsApiService.setEnabled(true); |
||||
|
} |
||||
|
} catch (Throwable e) { |
||||
|
log.error("Failed to start EDQS service", e); |
||||
|
} |
||||
|
}); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public void processSystemRequest(ToCoreEdqsRequest request) { |
||||
|
log.info("Processing system request {}", request); |
||||
|
if (request.getSyncRequest() != null) { |
||||
|
saveSyncState(EdqsSyncStatus.REQUESTED); |
||||
|
} |
||||
|
broadcast(request.toInternalMsg()); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public void processSystemMsg(ToCoreEdqsMsg msg) { |
||||
|
executor.submit(() -> { |
||||
|
log.info("Processing system msg {}", msg); |
||||
|
try { |
||||
|
if (msg.getApiEnabled() != null) { |
||||
|
edqsApiService.setEnabled(msg.getApiEnabled()); |
||||
|
} |
||||
|
|
||||
|
if (msg.getSyncRequest() != null) { |
||||
|
syncLock.lock(); |
||||
|
try { |
||||
|
EdqsSyncState syncState = getSyncState(); |
||||
|
if (syncState != null && syncState.getStatus() == EdqsSyncStatus.FINISHED) { |
||||
|
log.info("EDQS sync is already finished"); |
||||
|
return; |
||||
|
} |
||||
|
|
||||
|
saveSyncState(EdqsSyncStatus.STARTED); |
||||
|
edqsSyncService.sync(); |
||||
|
saveSyncState(EdqsSyncStatus.FINISHED); |
||||
|
|
||||
|
if (edqsApiService.isSupported()) |
||||
|
if (edqsApiService.isAutoEnable()) { |
||||
|
log.info("EDQS sync is finished, auto-enabling API"); |
||||
|
broadcast(ToCoreEdqsMsg.builder() |
||||
|
.apiEnabled(Boolean.TRUE) |
||||
|
.build()); |
||||
|
} else { |
||||
|
log.info("EDQS sync is finished, but leaving API disabled"); |
||||
|
} |
||||
|
} catch (Exception e) { |
||||
|
log.error("Failed to complete sync", e); |
||||
|
saveSyncState(EdqsSyncStatus.FAILED); |
||||
|
} finally { |
||||
|
syncLock.unlock(); |
||||
|
} |
||||
|
} |
||||
|
} catch (Throwable e) { |
||||
|
log.error("Failed to process msg {}", msg, e); |
||||
|
} |
||||
|
}); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public void onUpdate(TenantId tenantId, EntityId entityId, Object entity) { |
||||
|
EntityType entityType = entityId.getEntityType(); |
||||
|
ObjectType objectType = ObjectType.fromEntityType(entityType); |
||||
|
if (!isEdqsType(tenantId, objectType)) { |
||||
|
log.trace("[{}][{}] Ignoring update event, type {} not supported", tenantId, entityId, entityType); |
||||
|
return; |
||||
|
} |
||||
|
onUpdate(tenantId, objectType, edqsConverter.toEntity(entityType, entity)); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public void onUpdate(TenantId tenantId, ObjectType objectType, EdqsObject object) { |
||||
|
processEvent(tenantId, objectType, EdqsEventType.UPDATED, object); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public void onDelete(TenantId tenantId, EntityId entityId) { |
||||
|
EntityType entityType = entityId.getEntityType(); |
||||
|
ObjectType objectType = ObjectType.fromEntityType(entityType); |
||||
|
if (!isEdqsType(tenantId, objectType)) { |
||||
|
log.trace("[{}][{}] Ignoring deletion event, type {} not supported", tenantId, entityId, entityType); |
||||
|
return; |
||||
|
} |
||||
|
onDelete(tenantId, objectType, new Entity(entityType, entityId.getId(), Long.MAX_VALUE)); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public void onDelete(TenantId tenantId, ObjectType objectType, EdqsObject object) { |
||||
|
processEvent(tenantId, objectType, EdqsEventType.DELETED, object); |
||||
|
} |
||||
|
|
||||
|
protected void processEvent(TenantId tenantId, ObjectType objectType, EdqsEventType eventType, EdqsObject object) { |
||||
|
executor.submit(() -> { |
||||
|
try { |
||||
|
String key = object.key(); |
||||
|
Long version = object.version(); |
||||
|
EdqsEventMsg.Builder eventMsg = EdqsEventMsg.newBuilder() |
||||
|
.setKey(key) |
||||
|
.setObjectType(objectType.name()) |
||||
|
.setData(ByteString.copyFrom(edqsConverter.serialize(objectType, object))) |
||||
|
.setEventType(eventType.name()); |
||||
|
if (version != null) { |
||||
|
eventMsg.setVersion(version); |
||||
|
} |
||||
|
eventsProducer.send(tenantId, objectType, key, ToEdqsMsg.newBuilder() |
||||
|
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
||||
|
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
||||
|
.setTs(System.currentTimeMillis()) |
||||
|
.setEventMsg(eventMsg) |
||||
|
.build()); |
||||
|
} catch (Throwable e) { |
||||
|
log.error("[{}] Failed to push {} event for {} {}", tenantId, eventType, objectType, object, e); |
||||
|
} |
||||
|
}); |
||||
|
} |
||||
|
|
||||
|
private boolean isEdqsType(TenantId tenantId, ObjectType objectType) { |
||||
|
if (objectType == null) { |
||||
|
return false; |
||||
|
} |
||||
|
if (!tenantId.isSysTenantId()) { |
||||
|
return ObjectType.edqsTypes.contains(objectType); |
||||
|
} else { |
||||
|
return ObjectType.edqsSystemTypes.contains(objectType); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
private void broadcast(ToCoreEdqsMsg msg) { |
||||
|
clusterService.broadcastToCore(ToCoreNotificationMsg.newBuilder() |
||||
|
.setToEdqsCoreServiceMsg(ToEdqsCoreServiceMsg.newBuilder() |
||||
|
.setValue(ByteString.copyFrom(JacksonUtil.writeValueAsBytes(msg)))) |
||||
|
.build()); |
||||
|
} |
||||
|
|
||||
|
@SneakyThrows |
||||
|
private EdqsSyncState getSyncState() { |
||||
|
EdqsSyncState state = attributesService.find(TenantId.SYS_TENANT_ID, TenantId.SYS_TENANT_ID, AttributeScope.SERVER_SCOPE, "edqsSyncState").get(30, TimeUnit.SECONDS) |
||||
|
.flatMap(KvEntry::getJsonValue) |
||||
|
.map(value -> JacksonUtil.fromString(value, EdqsSyncState.class)) |
||||
|
.orElse(null); |
||||
|
log.info("EDQS sync state: {}", state); |
||||
|
return state; |
||||
|
} |
||||
|
|
||||
|
@SneakyThrows |
||||
|
private void saveSyncState(EdqsSyncStatus status) { |
||||
|
EdqsSyncState state = new EdqsSyncState(status); |
||||
|
log.info("New EDQS sync state: {}", state); |
||||
|
attributesService.save(TenantId.SYS_TENANT_ID, TenantId.SYS_TENANT_ID, AttributeScope.SERVER_SCOPE, new BaseAttributeKvEntry( |
||||
|
new JsonDataEntry("edqsSyncState", JacksonUtil.toString(state)), |
||||
|
System.currentTimeMillis())).get(30, TimeUnit.SECONDS); |
||||
|
} |
||||
|
|
||||
|
@PreDestroy |
||||
|
private void stop() { |
||||
|
executor.shutdown(); |
||||
|
eventsProducer.stop(); |
||||
|
} |
||||
|
|
||||
|
@Data |
||||
|
@AllArgsConstructor |
||||
|
@NoArgsConstructor |
||||
|
private static class EdqsSyncState { |
||||
|
private EdqsSyncStatus status; |
||||
|
} |
||||
|
|
||||
|
private enum EdqsSyncStatus { |
||||
|
REQUESTED, |
||||
|
STARTED, |
||||
|
FINISHED, |
||||
|
FAILED |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,61 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2025 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.edqs; |
||||
|
|
||||
|
import lombok.RequiredArgsConstructor; |
||||
|
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; |
||||
|
import org.springframework.stereotype.Service; |
||||
|
import org.springframework.transaction.event.TransactionalEventListener; |
||||
|
import org.thingsboard.server.common.data.ObjectType; |
||||
|
import org.thingsboard.server.common.data.audit.ActionType; |
||||
|
import org.thingsboard.server.common.msg.edqs.EdqsService; |
||||
|
import org.thingsboard.server.dao.eventsourcing.DeleteEntityEvent; |
||||
|
import org.thingsboard.server.dao.eventsourcing.RelationActionEvent; |
||||
|
import org.thingsboard.server.dao.eventsourcing.SaveEntityEvent; |
||||
|
|
||||
|
@Service |
||||
|
@RequiredArgsConstructor |
||||
|
@ConditionalOnProperty(value = "queue.edqs.sync.enabled", havingValue = "true") |
||||
|
public class EdqsListener { |
||||
|
|
||||
|
private final EdqsService edqsService; |
||||
|
|
||||
|
@TransactionalEventListener(fallbackExecution = true) |
||||
|
public void onUpdate(SaveEntityEvent<?> event) { |
||||
|
if (event.getEntityId() == null || event.getEntity() == null) { |
||||
|
return; |
||||
|
} |
||||
|
edqsService.onUpdate(event.getTenantId(), event.getEntityId(), event.getEntity()); |
||||
|
} |
||||
|
|
||||
|
@TransactionalEventListener(fallbackExecution = true) |
||||
|
public void onDelete(DeleteEntityEvent<?> event) { |
||||
|
if (event.getEntityId() == null) { |
||||
|
return; |
||||
|
} |
||||
|
edqsService.onDelete(event.getTenantId(), event.getEntityId()); |
||||
|
} |
||||
|
|
||||
|
@TransactionalEventListener(fallbackExecution = true) |
||||
|
public void handleEvent(RelationActionEvent relationEvent) { |
||||
|
if (relationEvent.getActionType() == ActionType.RELATION_ADD_OR_UPDATE) { |
||||
|
edqsService.onUpdate(relationEvent.getTenantId(), ObjectType.RELATION, relationEvent.getRelation()); |
||||
|
} else if (relationEvent.getActionType() == ActionType.RELATION_DELETED) { |
||||
|
edqsService.onDelete(relationEvent.getTenantId(), ObjectType.RELATION, relationEvent.getRelation()); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,284 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2025 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.edqs; |
||||
|
|
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.springframework.beans.factory.annotation.Autowired; |
||||
|
import org.springframework.beans.factory.annotation.Value; |
||||
|
import org.springframework.context.annotation.Lazy; |
||||
|
import org.thingsboard.server.common.data.AttributeScope; |
||||
|
import org.thingsboard.server.common.data.EntityType; |
||||
|
import org.thingsboard.server.common.data.ObjectType; |
||||
|
import org.thingsboard.server.common.data.edqs.AttributeKv; |
||||
|
import org.thingsboard.server.common.data.edqs.EdqsEventType; |
||||
|
import org.thingsboard.server.common.data.edqs.EdqsObject; |
||||
|
import org.thingsboard.server.common.data.edqs.Entity; |
||||
|
import org.thingsboard.server.common.data.edqs.LatestTsKv; |
||||
|
import org.thingsboard.server.common.data.edqs.fields.EntityFields; |
||||
|
import org.thingsboard.server.common.data.id.EntityId; |
||||
|
import org.thingsboard.server.common.data.id.EntityIdFactory; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
import org.thingsboard.server.common.data.page.PageDataIterable; |
||||
|
import org.thingsboard.server.common.data.relation.RelationTypeGroup; |
||||
|
import org.thingsboard.server.dao.Dao; |
||||
|
import org.thingsboard.server.dao.attributes.AttributesDao; |
||||
|
import org.thingsboard.server.dao.dictionary.KeyDictionaryDao; |
||||
|
import org.thingsboard.server.dao.entity.EntityDaoRegistry; |
||||
|
import org.thingsboard.server.dao.model.sql.AttributeKvEntity; |
||||
|
import org.thingsboard.server.dao.model.sql.RelationEntity; |
||||
|
import org.thingsboard.server.dao.model.sqlts.dictionary.KeyDictionaryEntry; |
||||
|
import org.thingsboard.server.dao.model.sqlts.latest.TsKvLatestEntity; |
||||
|
import org.thingsboard.server.dao.sql.relation.RelationRepository; |
||||
|
import org.thingsboard.server.dao.sqlts.latest.TsKvLatestRepository; |
||||
|
|
||||
|
import java.util.List; |
||||
|
import java.util.Map; |
||||
|
import java.util.UUID; |
||||
|
import java.util.concurrent.ConcurrentHashMap; |
||||
|
import java.util.concurrent.atomic.AtomicInteger; |
||||
|
|
||||
|
import static org.thingsboard.server.common.data.ObjectType.ATTRIBUTE_KV; |
||||
|
import static org.thingsboard.server.common.data.ObjectType.LATEST_TS_KV; |
||||
|
import static org.thingsboard.server.common.data.ObjectType.RELATION; |
||||
|
import static org.thingsboard.server.common.data.ObjectType.edqsTenantTypes; |
||||
|
|
||||
|
@Slf4j |
||||
|
public abstract class EdqsSyncService { |
||||
|
|
||||
|
@Value("${queue.edqs.sync.entity_batch_size:10000}") |
||||
|
private int entityBatchSize; |
||||
|
@Value("${queue.edqs.sync.ts_batch_size:10000}") |
||||
|
private int tsBatchSize; |
||||
|
@Autowired |
||||
|
private EntityDaoRegistry entityDaoRegistry; |
||||
|
@Autowired |
||||
|
private AttributesDao attributesDao; |
||||
|
@Autowired |
||||
|
private KeyDictionaryDao keyDictionaryDao; |
||||
|
@Autowired |
||||
|
private RelationRepository relationRepository; |
||||
|
@Autowired |
||||
|
private TsKvLatestRepository tsKvLatestRepository; |
||||
|
@Autowired |
||||
|
@Lazy |
||||
|
private DefaultEdqsService edqsService; |
||||
|
|
||||
|
private final ConcurrentHashMap<UUID, EntityIdInfo> entityInfoMap = new ConcurrentHashMap<>(); |
||||
|
private final ConcurrentHashMap<Integer, String> keys = new ConcurrentHashMap<>(); |
||||
|
|
||||
|
private final Map<ObjectType, AtomicInteger> counters = new ConcurrentHashMap<>(); |
||||
|
|
||||
|
public abstract boolean isSyncNeeded(); |
||||
|
|
||||
|
public void sync() { |
||||
|
log.info("Synchronizing data to EDQS"); |
||||
|
long startTs = System.currentTimeMillis(); |
||||
|
counters.clear(); |
||||
|
|
||||
|
syncTenantEntities(); |
||||
|
syncRelations(); |
||||
|
loadKeyDictionary(); |
||||
|
syncAttributes(); |
||||
|
syncLatestTimeseries(); |
||||
|
|
||||
|
counters.clear(); |
||||
|
log.info("Finishing synchronizing data to EDQS in {} ms", (System.currentTimeMillis() - startTs)); |
||||
|
} |
||||
|
|
||||
|
private void process(TenantId tenantId, ObjectType type, EdqsObject object) { |
||||
|
AtomicInteger counter = counters.computeIfAbsent(type, t -> new AtomicInteger()); |
||||
|
if (counter.incrementAndGet() % 10000 == 0) { |
||||
|
log.info("Processed {} {} objects", counter.get(), type); |
||||
|
} |
||||
|
edqsService.processEvent(tenantId, type, EdqsEventType.UPDATED, object); |
||||
|
} |
||||
|
|
||||
|
private void syncTenantEntities() { |
||||
|
for (ObjectType type : edqsTenantTypes) { |
||||
|
log.info("Synchronizing {} entities to EDQS", type); |
||||
|
long ts = System.currentTimeMillis(); |
||||
|
EntityType entityType = type.toEntityType(); |
||||
|
Dao<?> dao = entityDaoRegistry.getDao(entityType); |
||||
|
UUID lastId = UUID.fromString("00000000-0000-0000-0000-000000000000"); |
||||
|
while (true) { |
||||
|
var batch = dao.findNextBatch(lastId, entityBatchSize); |
||||
|
if (batch.isEmpty()) { |
||||
|
break; |
||||
|
} |
||||
|
for (EntityFields entityFields : batch) { |
||||
|
TenantId tenantId = TenantId.fromUUID(entityFields.getTenantId()); |
||||
|
entityInfoMap.put(entityFields.getId(), new EntityIdInfo(entityType, tenantId)); |
||||
|
process(tenantId, type, new Entity(entityType, entityFields)); |
||||
|
} |
||||
|
EntityFields lastRecord = batch.get(batch.size() - 1); |
||||
|
lastId = lastRecord.getId(); |
||||
|
} |
||||
|
log.info("Finished synchronizing {} entities to EDQS in {} ms", type, (System.currentTimeMillis() - ts)); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
private void syncRelations() { |
||||
|
log.info("Synchronizing relations to EDQS"); |
||||
|
long ts = System.currentTimeMillis(); |
||||
|
UUID lastFromEntityId = UUID.fromString("00000000-0000-0000-0000-000000000000"); |
||||
|
String lastFromEntityType = ""; |
||||
|
String lastRelationTypeGroup = ""; |
||||
|
String lastRelationType = ""; |
||||
|
UUID lastToEntityId = UUID.fromString("00000000-0000-0000-0000-000000000000"); |
||||
|
String lastToEntityType = ""; |
||||
|
|
||||
|
while (true) { |
||||
|
List<RelationEntity> batch = relationRepository.findNextBatch(lastFromEntityId, lastFromEntityType, lastRelationTypeGroup, |
||||
|
lastRelationType, lastToEntityId, lastToEntityType, entityBatchSize); |
||||
|
if (batch.isEmpty()) { |
||||
|
break; |
||||
|
} |
||||
|
processRelationBatch(batch); |
||||
|
|
||||
|
RelationEntity lastRecord = batch.get(batch.size() - 1); |
||||
|
lastFromEntityId = lastRecord.getFromId(); |
||||
|
lastFromEntityType = lastRecord.getFromType(); |
||||
|
lastRelationTypeGroup = lastRecord.getRelationTypeGroup(); |
||||
|
lastRelationType = lastRecord.getRelationType(); |
||||
|
lastToEntityId = lastRecord.getToId(); |
||||
|
lastToEntityType = lastRecord.getToType(); |
||||
|
} |
||||
|
log.info("Finished synchronizing relations to EDQS in {} ms", (System.currentTimeMillis() - ts)); |
||||
|
} |
||||
|
|
||||
|
private void processRelationBatch(List<RelationEntity> relations) { |
||||
|
for (RelationEntity relation : relations) { |
||||
|
if (RelationTypeGroup.COMMON.name().equals(relation.getRelationTypeGroup())) { |
||||
|
EntityIdInfo entityIdInfo = entityInfoMap.get(relation.getFromId()); |
||||
|
if (entityIdInfo != null) { |
||||
|
process(entityIdInfo.tenantId(), RELATION, relation.toData()); |
||||
|
} else { |
||||
|
log.info("Relation from id not found: {} ", relation); |
||||
|
} |
||||
|
} |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
private void loadKeyDictionary() { |
||||
|
log.info("Loading key dictionary"); |
||||
|
long ts = System.currentTimeMillis(); |
||||
|
var keyDictionaryEntries = new PageDataIterable<>(keyDictionaryDao::findAll, 10000); |
||||
|
for (KeyDictionaryEntry keyDictionaryEntry : keyDictionaryEntries) { |
||||
|
keys.put(keyDictionaryEntry.getKeyId(), keyDictionaryEntry.getKey()); |
||||
|
} |
||||
|
log.info("Finished loading key dictionary in {} ms", (System.currentTimeMillis() - ts)); |
||||
|
} |
||||
|
|
||||
|
private void syncAttributes() { |
||||
|
log.info("Synchronizing attributes to EDQS"); |
||||
|
long ts = System.currentTimeMillis(); |
||||
|
|
||||
|
UUID lastEntityId = UUID.fromString("00000000-0000-0000-0000-000000000000"); |
||||
|
int lastAttributeType = Integer.MIN_VALUE; |
||||
|
int lastAttributeKey = Integer.MIN_VALUE; |
||||
|
|
||||
|
while (true) { |
||||
|
List<AttributeKvEntity> batch = attributesDao.findNextBatch(lastEntityId, lastAttributeType, lastAttributeKey, tsBatchSize); |
||||
|
if (batch.isEmpty()) { |
||||
|
break; |
||||
|
} |
||||
|
processAttributeBatch(batch); |
||||
|
|
||||
|
AttributeKvEntity lastRecord = batch.get(batch.size() - 1); |
||||
|
lastEntityId = lastRecord.getId().getEntityId(); |
||||
|
lastAttributeType = lastRecord.getId().getAttributeType(); |
||||
|
lastAttributeKey = lastRecord.getId().getAttributeKey(); |
||||
|
} |
||||
|
log.info("Finished synchronizing attributes to EDQS in {} ms", (System.currentTimeMillis() - ts)); |
||||
|
} |
||||
|
|
||||
|
private void processAttributeBatch(List<AttributeKvEntity> batch) { |
||||
|
for (AttributeKvEntity attribute : batch) { |
||||
|
attribute.setStrKey(getStrKeyOrFetchFromDb(attribute.getId().getAttributeKey())); |
||||
|
UUID entityId = attribute.getId().getEntityId(); |
||||
|
EntityIdInfo entityIdInfo = entityInfoMap.get(entityId); |
||||
|
if (entityIdInfo == null) { |
||||
|
log.debug("Skipping attribute with entity UUID {} as it is not found in entityInfoMap", entityId); |
||||
|
continue; |
||||
|
} |
||||
|
AttributeKv attributeKv = new AttributeKv( |
||||
|
EntityIdFactory.getByTypeAndUuid(entityIdInfo.entityType(), entityId), |
||||
|
AttributeScope.valueOf(attribute.getId().getAttributeType()), |
||||
|
attribute.toData(), |
||||
|
attribute.getVersion()); |
||||
|
process(entityIdInfo.tenantId(), ATTRIBUTE_KV, attributeKv); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
private void syncLatestTimeseries() { |
||||
|
log.info("Synchronizing latest timeseries to EDQS"); |
||||
|
long ts = System.currentTimeMillis(); |
||||
|
UUID lastEntityId = UUID.fromString("00000000-0000-0000-0000-000000000000"); |
||||
|
int lastKey = Integer.MIN_VALUE; |
||||
|
|
||||
|
while (true) { |
||||
|
List<TsKvLatestEntity> batch = tsKvLatestRepository.findNextBatch(lastEntityId, lastKey, tsBatchSize); |
||||
|
if (batch.isEmpty()) { |
||||
|
break; |
||||
|
} |
||||
|
processTsKvLatestBatch(batch); |
||||
|
|
||||
|
TsKvLatestEntity lastRecord = batch.get(batch.size() - 1); |
||||
|
lastEntityId = lastRecord.getEntityId(); |
||||
|
lastKey = lastRecord.getKey(); |
||||
|
} |
||||
|
log.info("Finished synchronizing latest timeseries to EDQS in {} ms", (System.currentTimeMillis() - ts)); |
||||
|
} |
||||
|
|
||||
|
private void processTsKvLatestBatch(List<TsKvLatestEntity> tsKvLatestEntities) { |
||||
|
for (TsKvLatestEntity tsKvLatestEntity : tsKvLatestEntities) { |
||||
|
try { |
||||
|
String strKey = getStrKeyOrFetchFromDb(tsKvLatestEntity.getKey()); |
||||
|
if (strKey == null) { |
||||
|
log.debug("Skipping latest timeseries with key {} as it is not found in key dictionary", tsKvLatestEntity.getKey()); |
||||
|
continue; |
||||
|
} |
||||
|
tsKvLatestEntity.setStrKey(strKey); |
||||
|
UUID entityUuid = tsKvLatestEntity.getEntityId(); |
||||
|
EntityIdInfo entityIdInfo = entityInfoMap.get(entityUuid); |
||||
|
if (entityIdInfo != null) { |
||||
|
EntityId entityId = EntityIdFactory.getByTypeAndUuid(entityIdInfo.entityType(), entityUuid); |
||||
|
LatestTsKv latestTsKv = new LatestTsKv(entityId, tsKvLatestEntity.toData(), tsKvLatestEntity.getVersion()); |
||||
|
process(entityIdInfo.tenantId(), LATEST_TS_KV, latestTsKv); |
||||
|
} |
||||
|
} catch (Exception e) { |
||||
|
log.error("Failed to sync latest timeseries: {}", tsKvLatestEntity, e); |
||||
|
} |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
private String getStrKeyOrFetchFromDb(int key) { |
||||
|
String strKey = keys.get(key); |
||||
|
if (strKey != null) { |
||||
|
return strKey; |
||||
|
} else { |
||||
|
strKey = keyDictionaryDao.getKey(key); |
||||
|
if (strKey != null) { |
||||
|
keys.put(key, strKey); |
||||
|
} |
||||
|
} |
||||
|
return strKey; |
||||
|
} |
||||
|
|
||||
|
public record EntityIdInfo(EntityType entityType, TenantId tenantId) { |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,42 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2025 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.edqs; |
||||
|
|
||||
|
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; |
||||
|
import org.springframework.stereotype.Service; |
||||
|
import org.thingsboard.server.queue.edqs.EdqsQueue; |
||||
|
import org.thingsboard.server.queue.kafka.TbKafkaAdmin; |
||||
|
import org.thingsboard.server.queue.kafka.TbKafkaSettings; |
||||
|
|
||||
|
import java.util.Collections; |
||||
|
|
||||
|
@Service |
||||
|
@ConditionalOnExpression("'${queue.edqs.sync.enabled:true}' == 'true' && '${queue.type:null}' == 'kafka'") |
||||
|
public class KafkaEdqsSyncService extends EdqsSyncService { |
||||
|
|
||||
|
private final boolean syncNeeded; |
||||
|
|
||||
|
public KafkaEdqsSyncService(TbKafkaSettings kafkaSettings) { |
||||
|
TbKafkaAdmin kafkaAdmin = new TbKafkaAdmin(kafkaSettings, Collections.emptyMap()); |
||||
|
this.syncNeeded = kafkaAdmin.isTopicEmpty(EdqsQueue.EVENTS.getTopic()); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public boolean isSyncNeeded() { |
||||
|
return syncNeeded; |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,35 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2025 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.edqs; |
||||
|
|
||||
|
import lombok.RequiredArgsConstructor; |
||||
|
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; |
||||
|
import org.springframework.stereotype.Service; |
||||
|
import org.thingsboard.server.edqs.util.EdqsRocksDb; |
||||
|
|
||||
|
@Service |
||||
|
@RequiredArgsConstructor |
||||
|
@ConditionalOnExpression("'${queue.edqs.sync.enabled:true}' == 'true' && '${queue.type:null}' == 'in-memory'") |
||||
|
public class LocalEdqsSyncService extends EdqsSyncService { |
||||
|
|
||||
|
private final EdqsRocksDb db; |
||||
|
|
||||
|
@Override |
||||
|
public boolean isSyncNeeded() { |
||||
|
return db.isNew(); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,180 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2025 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.state; |
||||
|
|
||||
|
import lombok.RequiredArgsConstructor; |
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.springframework.stereotype.Service; |
||||
|
import org.thingsboard.rule.engine.api.DeviceStateManager; |
||||
|
import org.thingsboard.server.cluster.TbClusterService; |
||||
|
import org.thingsboard.server.common.data.id.DeviceId; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
import org.thingsboard.server.common.msg.queue.ServiceType; |
||||
|
import org.thingsboard.server.common.msg.queue.TbCallback; |
||||
|
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; |
||||
|
import org.thingsboard.server.gen.transport.TransportProtos; |
||||
|
import org.thingsboard.server.queue.common.SimpleTbQueueCallback; |
||||
|
import org.thingsboard.server.queue.discovery.PartitionService; |
||||
|
import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; |
||||
|
|
||||
|
import java.util.Optional; |
||||
|
import java.util.function.Consumer; |
||||
|
import java.util.function.Supplier; |
||||
|
|
||||
|
@Slf4j |
||||
|
@Service |
||||
|
@RequiredArgsConstructor |
||||
|
public class DefaultDeviceStateManager implements DeviceStateManager { |
||||
|
|
||||
|
private final TbServiceInfoProvider serviceInfoProvider; |
||||
|
private final PartitionService partitionService; |
||||
|
|
||||
|
private final Optional<DeviceStateService> deviceStateService; |
||||
|
private final TbClusterService clusterService; |
||||
|
|
||||
|
@Override |
||||
|
public void onDeviceConnect(TenantId tenantId, DeviceId deviceId, long connectTime, TbCallback callback) { |
||||
|
forwardToDeviceStateService(tenantId, deviceId, |
||||
|
deviceStateService -> { |
||||
|
log.debug("[{}][{}] Forwarding device connect event to local service. Connect time: [{}].", tenantId.getId(), deviceId.getId(), connectTime); |
||||
|
deviceStateService.onDeviceConnect(tenantId, deviceId, connectTime); |
||||
|
}, |
||||
|
() -> { |
||||
|
log.debug("[{}][{}] Sending device connect message to core. Connect time: [{}].", tenantId.getId(), deviceId.getId(), connectTime); |
||||
|
var deviceConnectMsg = TransportProtos.DeviceConnectProto.newBuilder() |
||||
|
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
||||
|
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
||||
|
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
||||
|
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
||||
|
.setLastConnectTime(connectTime) |
||||
|
.build(); |
||||
|
return TransportProtos.ToCoreMsg.newBuilder() |
||||
|
.setDeviceConnectMsg(deviceConnectMsg) |
||||
|
.build(); |
||||
|
}, callback); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public void onDeviceActivity(TenantId tenantId, DeviceId deviceId, long activityTime, TbCallback callback) { |
||||
|
forwardToDeviceStateService(tenantId, deviceId, |
||||
|
deviceStateService -> { |
||||
|
log.debug("[{}][{}] Forwarding device activity event to local service. Activity time: [{}].", tenantId.getId(), deviceId.getId(), activityTime); |
||||
|
deviceStateService.onDeviceActivity(tenantId, deviceId, activityTime); |
||||
|
}, |
||||
|
() -> { |
||||
|
log.debug("[{}][{}] Sending device activity message to core. Activity time: [{}].", tenantId.getId(), deviceId.getId(), activityTime); |
||||
|
var deviceActivityMsg = TransportProtos.DeviceActivityProto.newBuilder() |
||||
|
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
||||
|
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
||||
|
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
||||
|
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
||||
|
.setLastActivityTime(activityTime) |
||||
|
.build(); |
||||
|
return TransportProtos.ToCoreMsg.newBuilder() |
||||
|
.setDeviceActivityMsg(deviceActivityMsg) |
||||
|
.build(); |
||||
|
}, callback); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public void onDeviceDisconnect(TenantId tenantId, DeviceId deviceId, long disconnectTime, TbCallback callback) { |
||||
|
forwardToDeviceStateService(tenantId, deviceId, |
||||
|
deviceStateService -> { |
||||
|
log.debug("[{}][{}] Forwarding device disconnect event to local service. Disconnect time: [{}].", tenantId.getId(), deviceId.getId(), disconnectTime); |
||||
|
deviceStateService.onDeviceDisconnect(tenantId, deviceId, disconnectTime); |
||||
|
}, |
||||
|
() -> { |
||||
|
log.debug("[{}][{}] Sending device disconnect message to core. Disconnect time: [{}].", tenantId.getId(), deviceId.getId(), disconnectTime); |
||||
|
var deviceDisconnectMsg = TransportProtos.DeviceDisconnectProto.newBuilder() |
||||
|
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
||||
|
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
||||
|
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
||||
|
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
||||
|
.setLastDisconnectTime(disconnectTime) |
||||
|
.build(); |
||||
|
return TransportProtos.ToCoreMsg.newBuilder() |
||||
|
.setDeviceDisconnectMsg(deviceDisconnectMsg) |
||||
|
.build(); |
||||
|
}, callback); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public void onDeviceInactivity(TenantId tenantId, DeviceId deviceId, long inactivityTime, TbCallback callback) { |
||||
|
forwardToDeviceStateService(tenantId, deviceId, |
||||
|
deviceStateService -> { |
||||
|
log.debug("[{}][{}] Forwarding device inactivity event to local service. Inactivity time: [{}].", tenantId.getId(), deviceId.getId(), inactivityTime); |
||||
|
deviceStateService.onDeviceInactivity(tenantId, deviceId, inactivityTime); |
||||
|
}, |
||||
|
() -> { |
||||
|
log.debug("[{}][{}] Sending device inactivity message to core. Inactivity time: [{}].", tenantId.getId(), deviceId.getId(), inactivityTime); |
||||
|
var deviceInactivityMsg = TransportProtos.DeviceInactivityProto.newBuilder() |
||||
|
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
||||
|
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
||||
|
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
||||
|
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
||||
|
.setLastInactivityTime(inactivityTime) |
||||
|
.build(); |
||||
|
return TransportProtos.ToCoreMsg.newBuilder() |
||||
|
.setDeviceInactivityMsg(deviceInactivityMsg) |
||||
|
.build(); |
||||
|
}, callback); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public void onDeviceInactivityTimeoutUpdate(TenantId tenantId, DeviceId deviceId, long inactivityTimeout, TbCallback callback) { |
||||
|
forwardToDeviceStateService(tenantId, deviceId, |
||||
|
deviceStateService -> { |
||||
|
log.debug("[{}][{}] Forwarding device inactivity timeout update to local service. Updated inactivity timeout: [{}].", tenantId.getId(), deviceId.getId(), inactivityTimeout); |
||||
|
deviceStateService.onDeviceInactivityTimeoutUpdate(tenantId, deviceId, inactivityTimeout); |
||||
|
}, |
||||
|
() -> { |
||||
|
log.debug("[{}][{}] Sending device inactivity timeout update message to core. Updated inactivity timeout: [{}].", tenantId.getId(), deviceId.getId(), inactivityTimeout); |
||||
|
var deviceInactivityTimeoutUpdateMsg = TransportProtos.DeviceInactivityTimeoutUpdateProto.newBuilder() |
||||
|
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
||||
|
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
||||
|
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
||||
|
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
||||
|
.setInactivityTimeout(inactivityTimeout) |
||||
|
.build(); |
||||
|
return TransportProtos.ToCoreMsg.newBuilder() |
||||
|
.setDeviceInactivityTimeoutUpdateMsg(deviceInactivityTimeoutUpdateMsg) |
||||
|
.build(); |
||||
|
}, callback); |
||||
|
} |
||||
|
|
||||
|
private void forwardToDeviceStateService( |
||||
|
TenantId tenantId, DeviceId deviceId, |
||||
|
Consumer<DeviceStateService> toDeviceStateService, |
||||
|
Supplier<TransportProtos.ToCoreMsg> toCore, |
||||
|
TbCallback callback |
||||
|
) { |
||||
|
TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, tenantId, deviceId); |
||||
|
if (serviceInfoProvider.isService(ServiceType.TB_CORE) && tpi.isMyPartition() && deviceStateService.isPresent()) { |
||||
|
try { |
||||
|
toDeviceStateService.accept(deviceStateService.get()); |
||||
|
} catch (Exception e) { |
||||
|
log.error("[{}][{}] Failed to process device connectivity event.", tenantId.getId(), deviceId.getId(), e); |
||||
|
callback.onFailure(e); |
||||
|
return; |
||||
|
} |
||||
|
callback.onSuccess(); |
||||
|
} else { |
||||
|
TransportProtos.ToCoreMsg toCoreMsg = toCore.get(); |
||||
|
clusterService.pushMsgToCore(tpi, deviceId.getId(), toCoreMsg, new SimpleTbQueueCallback(__ -> callback.onSuccess(), callback::onFailure)); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -1,196 +0,0 @@ |
|||||
/** |
|
||||
* Copyright © 2016-2025 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.state; |
|
||||
|
|
||||
import lombok.Getter; |
|
||||
import lombok.extern.slf4j.Slf4j; |
|
||||
import org.springframework.stereotype.Service; |
|
||||
import org.thingsboard.rule.engine.api.RuleEngineDeviceStateManager; |
|
||||
import org.thingsboard.server.cluster.TbClusterService; |
|
||||
import org.thingsboard.server.common.data.id.DeviceId; |
|
||||
import org.thingsboard.server.common.data.id.TenantId; |
|
||||
import org.thingsboard.server.common.msg.queue.ServiceType; |
|
||||
import org.thingsboard.server.common.msg.queue.TbCallback; |
|
||||
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; |
|
||||
import org.thingsboard.server.gen.transport.TransportProtos; |
|
||||
import org.thingsboard.server.queue.common.SimpleTbQueueCallback; |
|
||||
import org.thingsboard.server.queue.discovery.PartitionService; |
|
||||
import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; |
|
||||
import org.thingsboard.server.queue.util.TbRuleEngineComponent; |
|
||||
|
|
||||
import java.util.Optional; |
|
||||
import java.util.UUID; |
|
||||
|
|
||||
@Slf4j |
|
||||
@Service |
|
||||
@TbRuleEngineComponent |
|
||||
public class DefaultRuleEngineDeviceStateManager implements RuleEngineDeviceStateManager { |
|
||||
|
|
||||
private final TbServiceInfoProvider serviceInfoProvider; |
|
||||
private final PartitionService partitionService; |
|
||||
|
|
||||
private final Optional<DeviceStateService> deviceStateService; |
|
||||
private final TbClusterService clusterService; |
|
||||
|
|
||||
public DefaultRuleEngineDeviceStateManager( |
|
||||
TbServiceInfoProvider serviceInfoProvider, PartitionService partitionService, |
|
||||
Optional<DeviceStateService> deviceStateServiceOptional, TbClusterService clusterService |
|
||||
) { |
|
||||
this.serviceInfoProvider = serviceInfoProvider; |
|
||||
this.partitionService = partitionService; |
|
||||
this.deviceStateService = deviceStateServiceOptional; |
|
||||
this.clusterService = clusterService; |
|
||||
} |
|
||||
|
|
||||
@Getter |
|
||||
private abstract static class ConnectivityEventInfo { |
|
||||
|
|
||||
private final TenantId tenantId; |
|
||||
private final DeviceId deviceId; |
|
||||
private final long eventTime; |
|
||||
|
|
||||
private ConnectivityEventInfo(TenantId tenantId, DeviceId deviceId, long eventTime) { |
|
||||
this.tenantId = tenantId; |
|
||||
this.deviceId = deviceId; |
|
||||
this.eventTime = eventTime; |
|
||||
} |
|
||||
|
|
||||
abstract void forwardToLocalService(); |
|
||||
|
|
||||
abstract TransportProtos.ToCoreMsg toQueueMsg(); |
|
||||
|
|
||||
} |
|
||||
|
|
||||
@Override |
|
||||
public void onDeviceConnect(TenantId tenantId, DeviceId deviceId, long connectTime, TbCallback callback) { |
|
||||
routeEvent(new ConnectivityEventInfo(tenantId, deviceId, connectTime) { |
|
||||
@Override |
|
||||
void forwardToLocalService() { |
|
||||
deviceStateService.ifPresent(service -> service.onDeviceConnect(tenantId, deviceId, connectTime)); |
|
||||
} |
|
||||
|
|
||||
@Override |
|
||||
TransportProtos.ToCoreMsg toQueueMsg() { |
|
||||
var deviceConnectMsg = TransportProtos.DeviceConnectProto.newBuilder() |
|
||||
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
|
||||
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
|
||||
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
|
||||
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
|
||||
.setLastConnectTime(connectTime) |
|
||||
.build(); |
|
||||
return TransportProtos.ToCoreMsg.newBuilder() |
|
||||
.setDeviceConnectMsg(deviceConnectMsg) |
|
||||
.build(); |
|
||||
} |
|
||||
}, callback); |
|
||||
} |
|
||||
|
|
||||
@Override |
|
||||
public void onDeviceActivity(TenantId tenantId, DeviceId deviceId, long activityTime, TbCallback callback) { |
|
||||
routeEvent(new ConnectivityEventInfo(tenantId, deviceId, activityTime) { |
|
||||
@Override |
|
||||
void forwardToLocalService() { |
|
||||
deviceStateService.ifPresent(service -> service.onDeviceActivity(tenantId, deviceId, activityTime)); |
|
||||
} |
|
||||
|
|
||||
@Override |
|
||||
TransportProtos.ToCoreMsg toQueueMsg() { |
|
||||
var deviceActivityMsg = TransportProtos.DeviceActivityProto.newBuilder() |
|
||||
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
|
||||
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
|
||||
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
|
||||
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
|
||||
.setLastActivityTime(activityTime) |
|
||||
.build(); |
|
||||
return TransportProtos.ToCoreMsg.newBuilder() |
|
||||
.setDeviceActivityMsg(deviceActivityMsg) |
|
||||
.build(); |
|
||||
} |
|
||||
}, callback); |
|
||||
} |
|
||||
|
|
||||
@Override |
|
||||
public void onDeviceDisconnect(TenantId tenantId, DeviceId deviceId, long disconnectTime, TbCallback callback) { |
|
||||
routeEvent(new ConnectivityEventInfo(tenantId, deviceId, disconnectTime) { |
|
||||
@Override |
|
||||
void forwardToLocalService() { |
|
||||
deviceStateService.ifPresent(service -> service.onDeviceDisconnect(tenantId, deviceId, disconnectTime)); |
|
||||
} |
|
||||
|
|
||||
@Override |
|
||||
TransportProtos.ToCoreMsg toQueueMsg() { |
|
||||
var deviceDisconnectMsg = TransportProtos.DeviceDisconnectProto.newBuilder() |
|
||||
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
|
||||
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
|
||||
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
|
||||
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
|
||||
.setLastDisconnectTime(disconnectTime) |
|
||||
.build(); |
|
||||
return TransportProtos.ToCoreMsg.newBuilder() |
|
||||
.setDeviceDisconnectMsg(deviceDisconnectMsg) |
|
||||
.build(); |
|
||||
} |
|
||||
}, callback); |
|
||||
} |
|
||||
|
|
||||
@Override |
|
||||
public void onDeviceInactivity(TenantId tenantId, DeviceId deviceId, long inactivityTime, TbCallback callback) { |
|
||||
routeEvent(new ConnectivityEventInfo(tenantId, deviceId, inactivityTime) { |
|
||||
@Override |
|
||||
void forwardToLocalService() { |
|
||||
deviceStateService.ifPresent(service -> service.onDeviceInactivity(tenantId, deviceId, inactivityTime)); |
|
||||
} |
|
||||
|
|
||||
@Override |
|
||||
TransportProtos.ToCoreMsg toQueueMsg() { |
|
||||
var deviceInactivityMsg = TransportProtos.DeviceInactivityProto.newBuilder() |
|
||||
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
|
||||
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
|
||||
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
|
||||
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
|
||||
.setLastInactivityTime(inactivityTime) |
|
||||
.build(); |
|
||||
return TransportProtos.ToCoreMsg.newBuilder() |
|
||||
.setDeviceInactivityMsg(deviceInactivityMsg) |
|
||||
.build(); |
|
||||
} |
|
||||
}, callback); |
|
||||
} |
|
||||
|
|
||||
private void routeEvent(ConnectivityEventInfo eventInfo, TbCallback callback) { |
|
||||
var tenantId = eventInfo.getTenantId(); |
|
||||
var deviceId = eventInfo.getDeviceId(); |
|
||||
long eventTime = eventInfo.getEventTime(); |
|
||||
|
|
||||
TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, tenantId, deviceId); |
|
||||
if (serviceInfoProvider.isService(ServiceType.TB_CORE) && tpi.isMyPartition() && deviceStateService.isPresent()) { |
|
||||
log.debug("[{}][{}] Forwarding device connectivity event to local service. Event time: [{}].", tenantId.getId(), deviceId.getId(), eventTime); |
|
||||
try { |
|
||||
eventInfo.forwardToLocalService(); |
|
||||
} catch (Exception e) { |
|
||||
log.error("[{}][{}] Failed to process device connectivity event. Event time: [{}].", tenantId.getId(), deviceId.getId(), eventTime, e); |
|
||||
callback.onFailure(e); |
|
||||
return; |
|
||||
} |
|
||||
callback.onSuccess(); |
|
||||
} else { |
|
||||
TransportProtos.ToCoreMsg msg = eventInfo.toQueueMsg(); |
|
||||
log.debug("[{}][{}] Sending device connectivity message to core. Event time: [{}].", tenantId.getId(), deviceId.getId(), eventTime); |
|
||||
clusterService.pushMsgToCore(tpi, UUID.randomUUID(), msg, new SimpleTbQueueCallback(__ -> callback.onSuccess(), callback::onFailure)); |
|
||||
} |
|
||||
} |
|
||||
|
|
||||
} |
|
||||
@ -0,0 +1,72 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2025 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.controller; |
||||
|
|
||||
|
import org.junit.Before; |
||||
|
import org.springframework.beans.factory.annotation.Autowired; |
||||
|
import org.springframework.boot.test.mock.mockito.MockBean; |
||||
|
import org.springframework.test.context.TestPropertySource; |
||||
|
import org.thingsboard.server.common.data.page.PageData; |
||||
|
import org.thingsboard.server.common.data.query.EntityCountQuery; |
||||
|
import org.thingsboard.server.common.data.query.EntityData; |
||||
|
import org.thingsboard.server.common.data.query.EntityDataQuery; |
||||
|
import org.thingsboard.server.common.msg.edqs.EdqsApiService; |
||||
|
import org.thingsboard.server.dao.service.DaoSqlTest; |
||||
|
import org.thingsboard.server.edqs.state.EdqsStateService; |
||||
|
import org.thingsboard.server.edqs.util.EdqsRocksDb; |
||||
|
|
||||
|
import java.util.concurrent.TimeUnit; |
||||
|
|
||||
|
import static org.awaitility.Awaitility.await; |
||||
|
|
||||
|
@DaoSqlTest |
||||
|
@TestPropertySource(properties = { |
||||
|
// "queue.type=kafka", // uncomment to use Kafka
|
||||
|
// "queue.kafka.bootstrap.servers=10.7.1.254:9092",
|
||||
|
"queue.edqs.sync.enabled=true", |
||||
|
"queue.edqs.api.supported=true", |
||||
|
"queue.edqs.api.auto_enable=true", |
||||
|
"queue.edqs.mode=local" |
||||
|
}) |
||||
|
public class EdqsEntityQueryControllerTest extends EntityQueryControllerTest { |
||||
|
|
||||
|
@Autowired |
||||
|
private EdqsApiService edqsApiService; |
||||
|
|
||||
|
@Autowired |
||||
|
private EdqsStateService edqsStateService; |
||||
|
|
||||
|
@MockBean // so that we don't do backup for tests
|
||||
|
private EdqsRocksDb edqsRocksDb; |
||||
|
|
||||
|
@Before |
||||
|
public void before() { |
||||
|
await().atMost(TIMEOUT, TimeUnit.SECONDS).until(() -> edqsApiService.isEnabled() && edqsStateService.isReady()); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
protected PageData<EntityData> findByQueryAndCheck(EntityDataQuery query, int expectedResultSize) { |
||||
|
return await().atMost(TIMEOUT, TimeUnit.SECONDS).until(() -> findByQuery(query), |
||||
|
result -> result.getTotalElements() == expectedResultSize); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
protected Long countByQueryAndCheck(EntityCountQuery query, long expectedResult) { |
||||
|
return await().atMost(TIMEOUT, TimeUnit.SECONDS).until(() -> countByQuery(query), |
||||
|
result -> result == expectedResult); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,128 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2025 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.entitiy; |
||||
|
|
||||
|
import com.google.common.collect.Lists; |
||||
|
import org.junit.Before; |
||||
|
import org.junit.Test; |
||||
|
import org.springframework.beans.factory.annotation.Autowired; |
||||
|
import org.springframework.boot.test.mock.mockito.MockBean; |
||||
|
import org.springframework.test.context.TestPropertySource; |
||||
|
import org.thingsboard.server.common.data.EntityType; |
||||
|
import org.thingsboard.server.common.data.asset.Asset; |
||||
|
import org.thingsboard.server.common.data.id.CustomerId; |
||||
|
import org.thingsboard.server.common.data.id.IdBased; |
||||
|
import org.thingsboard.server.common.data.page.PageData; |
||||
|
import org.thingsboard.server.common.data.query.EntityCountQuery; |
||||
|
import org.thingsboard.server.common.data.query.EntityData; |
||||
|
import org.thingsboard.server.common.data.query.EntityDataQuery; |
||||
|
import org.thingsboard.server.common.data.query.EntityKeyType; |
||||
|
import org.thingsboard.server.common.data.query.RelationsQueryFilter; |
||||
|
import org.thingsboard.server.common.data.relation.EntitySearchDirection; |
||||
|
import org.thingsboard.server.common.data.relation.RelationEntityTypeFilter; |
||||
|
import org.thingsboard.server.common.msg.edqs.EdqsApiService; |
||||
|
import org.thingsboard.server.dao.service.DaoSqlTest; |
||||
|
import org.thingsboard.server.edqs.util.EdqsRocksDb; |
||||
|
|
||||
|
import java.util.ArrayList; |
||||
|
import java.util.Collections; |
||||
|
import java.util.HashMap; |
||||
|
import java.util.List; |
||||
|
import java.util.Map; |
||||
|
import java.util.UUID; |
||||
|
import java.util.concurrent.TimeUnit; |
||||
|
import java.util.stream.Collectors; |
||||
|
|
||||
|
import static org.awaitility.Awaitility.await; |
||||
|
|
||||
|
@DaoSqlTest |
||||
|
@TestPropertySource(properties = { |
||||
|
"queue.edqs.sync.enabled=true", |
||||
|
"queue.edqs.api.supported=true", |
||||
|
"queue.edqs.api.auto_enable=true", |
||||
|
"queue.edqs.mode=local" |
||||
|
}) |
||||
|
public class EdqsEntityServiceTest extends EntityServiceTest { |
||||
|
|
||||
|
@Autowired |
||||
|
private EdqsApiService edqsApiService; |
||||
|
|
||||
|
@MockBean |
||||
|
private EdqsRocksDb edqsRocksDb; |
||||
|
|
||||
|
@Before |
||||
|
public void beforeEach() { |
||||
|
await().atMost(TIMEOUT, TimeUnit.SECONDS).until(() -> edqsApiService.isEnabled()); |
||||
|
} |
||||
|
|
||||
|
// sql implementation has a bug with data duplication, edqs implementation returns correct value
|
||||
|
@Override |
||||
|
@Test |
||||
|
public void testCountHierarchicalEntitiesByMultiRootQuery() throws InterruptedException { |
||||
|
List<Asset> buildings = new ArrayList<>(); |
||||
|
List<Asset> apartments = new ArrayList<>(); |
||||
|
Map<String, Map<UUID, String>> entityNameByTypeMap = new HashMap<>(); |
||||
|
Map<UUID, UUID> childParentRelationMap = new HashMap<>(); |
||||
|
createMultiRootHierarchy(buildings, apartments, entityNameByTypeMap, childParentRelationMap); |
||||
|
|
||||
|
RelationsQueryFilter filter = new RelationsQueryFilter(); |
||||
|
filter.setMultiRoot(true); |
||||
|
filter.setMultiRootEntitiesType(EntityType.ASSET); |
||||
|
filter.setMultiRootEntityIds(buildings.stream().map(IdBased::getId).map(d -> d.getId().toString()).collect(Collectors.toSet())); |
||||
|
filter.setDirection(EntitySearchDirection.FROM); |
||||
|
|
||||
|
EntityCountQuery countQuery = new EntityCountQuery(filter); |
||||
|
countByQueryAndCheck(countQuery, 63); |
||||
|
|
||||
|
filter.setFilters(Collections.singletonList(new RelationEntityTypeFilter("AptToHeat", Collections.singletonList(EntityType.DEVICE)))); |
||||
|
countByQueryAndCheck(countQuery, 27); |
||||
|
|
||||
|
filter.setMultiRootEntitiesType(EntityType.ASSET); |
||||
|
filter.setMultiRootEntityIds(apartments.stream().map(IdBased::getId).map(d -> d.getId().toString()).collect(Collectors.toSet())); |
||||
|
filter.setDirection(EntitySearchDirection.TO); |
||||
|
filter.setFilters(Lists.newArrayList( |
||||
|
new RelationEntityTypeFilter("buildingToApt", Collections.singletonList(EntityType.ASSET)), |
||||
|
new RelationEntityTypeFilter("AptToEnergy", Collections.singletonList(EntityType.DEVICE)))); |
||||
|
countByQueryAndCheck(countQuery, 3); |
||||
|
|
||||
|
deviceService.deleteDevicesByTenantId(tenantId); |
||||
|
assetService.deleteAssetsByTenantId(tenantId); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
protected PageData<EntityData> findByQueryAndCheck(CustomerId customerId, EntityDataQuery query, long expectedResultSize) { |
||||
|
return await().atMost(TIMEOUT, TimeUnit.SECONDS).until(() -> findByQuery(customerId, query), |
||||
|
result -> result.getTotalElements() == expectedResultSize); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
protected List<EntityData> findByQueryAndCheckTelemetry(EntityDataQuery query, EntityKeyType entityKeyType, String key, List<String> expectedTelemetries) { |
||||
|
return await().atMost(TIMEOUT, TimeUnit.SECONDS).until(() -> findEntitiesTelemetry(query, entityKeyType, key, expectedTelemetries), |
||||
|
loadedEntities -> loadedEntities.stream().map(entityData -> entityData.getLatest().get(entityKeyType).get(key).getValue()).toList().containsAll(expectedTelemetries)); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
protected long countByQueryAndCheck(EntityCountQuery countQuery, int expectedResult) { |
||||
|
return countByQueryAndCheck(new CustomerId(CustomerId.NULL_UUID), countQuery, expectedResult); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
protected long countByQueryAndCheck(CustomerId customerId, EntityCountQuery query, int expectedResult) { |
||||
|
return await().atMost(TIMEOUT, TimeUnit.SECONDS).until(() -> countByQuery(customerId, query), |
||||
|
result -> result == expectedResult); |
||||
|
} |
||||
|
|
||||
|
} |
||||
File diff suppressed because it is too large
@ -0,0 +1,89 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2025 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 java.util.EnumSet; |
||||
|
import java.util.List; |
||||
|
import java.util.Set; |
||||
|
|
||||
|
public enum ObjectType { |
||||
|
TENANT, |
||||
|
TENANT_PROFILE, |
||||
|
CUSTOMER, |
||||
|
QUEUE, |
||||
|
RPC, |
||||
|
RULE_CHAIN, |
||||
|
OTA_PACKAGE, |
||||
|
RESOURCE, |
||||
|
EVENT, |
||||
|
RULE_NODE, |
||||
|
USER, |
||||
|
EDGE, |
||||
|
WIDGETS_BUNDLE, |
||||
|
WIDGET_TYPE, |
||||
|
DASHBOARD, |
||||
|
DEVICE_PROFILE, |
||||
|
DEVICE, |
||||
|
DEVICE_CREDENTIALS, |
||||
|
ASSET_PROFILE, |
||||
|
ASSET, |
||||
|
ENTITY_VIEW, |
||||
|
ALARM, |
||||
|
ENTITY_ALARM, |
||||
|
OAUTH2_CLIENT, |
||||
|
OAUTH2_DOMAIN, |
||||
|
OAUTH2_MOBILE, |
||||
|
USER_SETTINGS, |
||||
|
NOTIFICATION_TARGET, |
||||
|
NOTIFICATION_TEMPLATE, |
||||
|
NOTIFICATION_RULE, |
||||
|
ALARM_COMMENT, |
||||
|
API_USAGE_STATE, |
||||
|
QUEUE_STATS, |
||||
|
|
||||
|
AUDIT_LOG, |
||||
|
RELATION, |
||||
|
ATTRIBUTE_KV, |
||||
|
LATEST_TS_KV; |
||||
|
|
||||
|
public static final Set<ObjectType> edqsTenantTypes = EnumSet.of( |
||||
|
TENANT, CUSTOMER, DEVICE_PROFILE, DEVICE, ASSET_PROFILE, ASSET, EDGE, ENTITY_VIEW, USER, DASHBOARD, |
||||
|
RULE_CHAIN, WIDGET_TYPE, WIDGETS_BUNDLE, API_USAGE_STATE, QUEUE_STATS |
||||
|
); |
||||
|
public static final Set<ObjectType> edqsTypes = EnumSet.copyOf(edqsTenantTypes); |
||||
|
public static final Set<ObjectType> edqsSystemTypes = EnumSet.of(TENANT, USER, DASHBOARD, |
||||
|
API_USAGE_STATE, ATTRIBUTE_KV, LATEST_TS_KV); |
||||
|
public static final Set<ObjectType> unversionedTypes = EnumSet.of( |
||||
|
QUEUE_STATS // created once, never updated
|
||||
|
); |
||||
|
|
||||
|
static { |
||||
|
edqsTypes.addAll(List.of(RELATION, ATTRIBUTE_KV, LATEST_TS_KV)); |
||||
|
} |
||||
|
|
||||
|
public EntityType toEntityType() { |
||||
|
return EntityType.valueOf(name()); |
||||
|
} |
||||
|
|
||||
|
public static ObjectType fromEntityType(EntityType entityType) { |
||||
|
try { |
||||
|
return ObjectType.valueOf(entityType.name()); |
||||
|
} catch (Exception e) { |
||||
|
return null; |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,75 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2025 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.edqs; |
||||
|
|
||||
|
import lombok.AllArgsConstructor; |
||||
|
import lombok.Builder; |
||||
|
import lombok.Data; |
||||
|
import lombok.NoArgsConstructor; |
||||
|
import org.thingsboard.server.common.data.AttributeScope; |
||||
|
import org.thingsboard.server.common.data.ObjectType; |
||||
|
import org.thingsboard.server.common.data.id.EntityId; |
||||
|
import org.thingsboard.server.common.data.kv.AttributeKvEntry; |
||||
|
import org.thingsboard.server.common.data.kv.KvEntry; |
||||
|
|
||||
|
@Data |
||||
|
@AllArgsConstructor |
||||
|
@NoArgsConstructor |
||||
|
@Builder |
||||
|
public class AttributeKv implements EdqsObject { |
||||
|
|
||||
|
private EntityId entityId; |
||||
|
private AttributeScope scope; |
||||
|
private String key; |
||||
|
private Long version; |
||||
|
|
||||
|
private DataPoint dataPoint; // optional (on deletion)
|
||||
|
|
||||
|
private Long lastUpdateTs; // only for serialization
|
||||
|
private KvEntry value; // only for serialization
|
||||
|
|
||||
|
public AttributeKv(EntityId entityId, AttributeScope scope, AttributeKvEntry attributeKvEntry, long version) { |
||||
|
this.entityId = entityId; |
||||
|
this.scope = scope; |
||||
|
this.key = attributeKvEntry.getKey(); |
||||
|
this.version = version; |
||||
|
this.lastUpdateTs = attributeKvEntry.getLastUpdateTs(); |
||||
|
this.value = attributeKvEntry; |
||||
|
} |
||||
|
|
||||
|
public AttributeKv(EntityId entityId, AttributeScope scope, String key, long version) { |
||||
|
this.entityId = entityId; |
||||
|
this.scope = scope; |
||||
|
this.key = key; |
||||
|
this.version = version; |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public String key() { |
||||
|
return "a_" + entityId + "_" + scope + "_" + key; |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public Long version() { |
||||
|
return version; |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public ObjectType type() { |
||||
|
return ObjectType.ATTRIBUTE_KV; |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,40 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2025 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.edqs; |
||||
|
|
||||
|
import org.thingsboard.server.common.data.kv.DataType; |
||||
|
|
||||
|
public interface DataPoint { |
||||
|
|
||||
|
String NOT_SUPPORTED = "Not supported!"; |
||||
|
|
||||
|
long getTs(); |
||||
|
|
||||
|
DataType getType(); |
||||
|
|
||||
|
String getStr(); |
||||
|
|
||||
|
long getLong(); |
||||
|
|
||||
|
double getDouble(); |
||||
|
|
||||
|
boolean getBool(); |
||||
|
|
||||
|
String getJson(); |
||||
|
|
||||
|
String valueToString(); |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,34 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2025 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.edqs; |
||||
|
|
||||
|
import lombok.AllArgsConstructor; |
||||
|
import lombok.Builder; |
||||
|
import lombok.Data; |
||||
|
import org.thingsboard.server.common.data.ObjectType; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
|
||||
|
@Data |
||||
|
@AllArgsConstructor |
||||
|
@Builder |
||||
|
public class EdqsEvent { |
||||
|
|
||||
|
private final TenantId tenantId; |
||||
|
private final ObjectType objectType; |
||||
|
private final EdqsEventType eventType; |
||||
|
private final EdqsObject object; |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,21 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2025 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.edqs; |
||||
|
|
||||
|
public enum EdqsEventType { |
||||
|
UPDATED, |
||||
|
DELETED |
||||
|
} |
||||
@ -0,0 +1,32 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2025 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.edqs; |
||||
|
|
||||
|
import com.fasterxml.jackson.annotation.JsonIgnore; |
||||
|
import org.thingsboard.server.common.data.ObjectType; |
||||
|
|
||||
|
public interface EdqsObject { |
||||
|
|
||||
|
@JsonIgnore |
||||
|
String key(); |
||||
|
|
||||
|
@JsonIgnore |
||||
|
Long version(); |
||||
|
|
||||
|
@JsonIgnore |
||||
|
ObjectType type(); |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,24 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2025 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.edqs; |
||||
|
|
||||
|
import com.fasterxml.jackson.annotation.JsonIgnoreProperties; |
||||
|
import lombok.Data; |
||||
|
|
||||
|
@Data |
||||
|
@JsonIgnoreProperties |
||||
|
public class EdqsSyncRequest { |
||||
|
} |
||||
Some files were not shown because too many files changed in this diff
Loading…
Reference in new issue