|
|
|
@ -25,6 +25,7 @@ import lombok.extern.slf4j.Slf4j; |
|
|
|
import org.checkerframework.checker.nullness.qual.Nullable; |
|
|
|
import org.springframework.beans.factory.annotation.Autowired; |
|
|
|
import org.springframework.stereotype.Service; |
|
|
|
import org.thingsboard.server.common.data.AdminSettings; |
|
|
|
import org.thingsboard.server.common.data.DashboardInfo; |
|
|
|
import org.thingsboard.server.common.data.DataConstants; |
|
|
|
import org.thingsboard.server.common.data.Device; |
|
|
|
@ -54,6 +55,8 @@ import org.thingsboard.server.common.data.relation.EntityRelationsQuery; |
|
|
|
import org.thingsboard.server.common.data.relation.EntitySearchDirection; |
|
|
|
import org.thingsboard.server.common.data.relation.RelationsSearchParameters; |
|
|
|
import org.thingsboard.server.common.data.rule.RuleChain; |
|
|
|
import org.thingsboard.server.common.data.widget.WidgetType; |
|
|
|
import org.thingsboard.server.common.data.widget.WidgetsBundle; |
|
|
|
import org.thingsboard.server.dao.asset.AssetService; |
|
|
|
import org.thingsboard.server.dao.attributes.AttributesService; |
|
|
|
import org.thingsboard.server.dao.dashboard.DashboardService; |
|
|
|
@ -62,7 +65,10 @@ import org.thingsboard.server.dao.edge.EdgeEventService; |
|
|
|
import org.thingsboard.server.dao.entityview.EntityViewService; |
|
|
|
import org.thingsboard.server.dao.relation.RelationService; |
|
|
|
import org.thingsboard.server.dao.rule.RuleChainService; |
|
|
|
import org.thingsboard.server.dao.settings.AdminSettingsService; |
|
|
|
import org.thingsboard.server.dao.user.UserService; |
|
|
|
import org.thingsboard.server.dao.widget.WidgetTypeService; |
|
|
|
import org.thingsboard.server.dao.widget.WidgetsBundleService; |
|
|
|
import org.thingsboard.server.gen.edge.AttributesRequestMsg; |
|
|
|
import org.thingsboard.server.gen.edge.DeviceCredentialsRequestMsg; |
|
|
|
import org.thingsboard.server.gen.edge.RelationRequestMsg; |
|
|
|
@ -109,6 +115,15 @@ public class DefaultSyncEdgeService implements SyncEdgeService { |
|
|
|
@Autowired |
|
|
|
private UserService userService; |
|
|
|
|
|
|
|
@Autowired |
|
|
|
private WidgetsBundleService widgetsBundleService; |
|
|
|
|
|
|
|
@Autowired |
|
|
|
private WidgetTypeService widgetTypeService; |
|
|
|
|
|
|
|
@Autowired |
|
|
|
private AdminSettingsService adminSettingsService; |
|
|
|
|
|
|
|
@Autowired |
|
|
|
private DbCallbackExecutorService dbCallbackExecutorService; |
|
|
|
|
|
|
|
@ -121,6 +136,8 @@ public class DefaultSyncEdgeService implements SyncEdgeService { |
|
|
|
syncAssets(edge); |
|
|
|
syncEntityViews(edge); |
|
|
|
syncDashboards(edge); |
|
|
|
syncWidgetsBundleAndWidgetTypes(edge); |
|
|
|
syncAdminSettings(edge); |
|
|
|
} catch (Exception e) { |
|
|
|
log.error("Exception during sync process", e); |
|
|
|
} |
|
|
|
@ -253,6 +270,7 @@ public class DefaultSyncEdgeService implements SyncEdgeService { |
|
|
|
TextPageData<User> pageData = userService.findTenantAdmins(edge.getTenantId(), new TextPageLink(Integer.MAX_VALUE)); |
|
|
|
pushUsersToEdge(pageData, edge); |
|
|
|
if (edge.getCustomerId() != null && !EntityId.NULL_UUID.equals(edge.getCustomerId().getId())) { |
|
|
|
saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.CUSTOMER, ActionType.ADDED, edge.getCustomerId(), null); |
|
|
|
pageData = userService.findCustomerUsers(edge.getTenantId(), edge.getCustomerId(), new TextPageLink(Integer.MAX_VALUE)); |
|
|
|
pushUsersToEdge(pageData, edge); |
|
|
|
} |
|
|
|
@ -261,6 +279,33 @@ public class DefaultSyncEdgeService implements SyncEdgeService { |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
private void syncWidgetsBundleAndWidgetTypes(Edge edge) { |
|
|
|
List<WidgetsBundle> widgetsBundlesToPush = new ArrayList<>(); |
|
|
|
List<WidgetType> widgetTypesToPush = new ArrayList<>(); |
|
|
|
widgetsBundlesToPush.addAll(widgetsBundleService.findAllTenantWidgetsBundlesByTenantId(edge.getTenantId())); |
|
|
|
widgetsBundlesToPush.addAll(widgetsBundleService.findSystemWidgetsBundles(edge.getTenantId())); |
|
|
|
try { |
|
|
|
for (WidgetsBundle widgetsBundle: widgetsBundlesToPush) { |
|
|
|
saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.WIDGETS_BUNDLE, ActionType.ADDED, widgetsBundle.getId(), null); |
|
|
|
widgetTypesToPush.addAll(widgetTypeService.findWidgetTypesByTenantIdAndBundleAlias(widgetsBundle.getTenantId(), widgetsBundle.getAlias())); |
|
|
|
} |
|
|
|
for (WidgetType widgetType: widgetTypesToPush) { |
|
|
|
saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.WIDGET_TYPE, ActionType.ADDED, widgetType.getId(), null); |
|
|
|
} |
|
|
|
} catch (Exception e) { |
|
|
|
log.error("Exception during loading widgets bundle(s) and widget type(s) on sync!", e); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
private void syncAdminSettings(Edge edge) { |
|
|
|
try { |
|
|
|
AdminSettings mailSettings = adminSettingsService.findAdminSettingsByKey(TenantId.SYS_TENANT_ID, "mail"); |
|
|
|
saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.ADMIN_SETTINGS, ActionType.UPDATED, null, mapper.valueToTree(mailSettings)); |
|
|
|
} catch (Exception e) { |
|
|
|
log.error("Can't load admin settings", e); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
private void pushUsersToEdge(TextPageData<User> pageData, Edge edge) { |
|
|
|
if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) { |
|
|
|
log.trace("[{}] [{}] user(s) are going to be pushed to edge.", edge.getId(), pageData.getData().size()); |
|
|
|
@ -271,64 +316,62 @@ public class DefaultSyncEdgeService implements SyncEdgeService { |
|
|
|
} |
|
|
|
|
|
|
|
@Override |
|
|
|
public void processRuleChainMetadataRequestMsg(Edge edge, RuleChainMetadataRequestMsg ruleChainMetadataRequestMsg) { |
|
|
|
public ListenableFuture<Void> processRuleChainMetadataRequestMsg(Edge edge, RuleChainMetadataRequestMsg ruleChainMetadataRequestMsg) { |
|
|
|
if (ruleChainMetadataRequestMsg.getRuleChainIdMSB() != 0 && ruleChainMetadataRequestMsg.getRuleChainIdLSB() != 0) { |
|
|
|
RuleChainId ruleChainId = new RuleChainId(new UUID(ruleChainMetadataRequestMsg.getRuleChainIdMSB(), ruleChainMetadataRequestMsg.getRuleChainIdLSB())); |
|
|
|
saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.RULE_CHAIN_METADATA, ActionType.ADDED, ruleChainId, null); |
|
|
|
ListenableFuture<EdgeEvent> future = saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.RULE_CHAIN_METADATA, ActionType.ADDED, ruleChainId, null); |
|
|
|
return Futures.transform(future, edgeEvent -> null, dbCallbackExecutorService); |
|
|
|
} |
|
|
|
return Futures.immediateFuture(null); |
|
|
|
} |
|
|
|
|
|
|
|
@Override |
|
|
|
public void processAttributesRequestMsg(Edge edge, AttributesRequestMsg attributesRequestMsg) { |
|
|
|
public ListenableFuture<Void> processAttributesRequestMsg(Edge edge, AttributesRequestMsg attributesRequestMsg) { |
|
|
|
EntityId entityId = EntityIdFactory.getByTypeAndUuid( |
|
|
|
EntityType.valueOf(attributesRequestMsg.getEntityType()), |
|
|
|
new UUID(attributesRequestMsg.getEntityIdMSB(), attributesRequestMsg.getEntityIdLSB())); |
|
|
|
final EdgeEventType edgeEventType = getEdgeQueueTypeByEntityType(entityId.getEntityType()); |
|
|
|
if (edgeEventType != null) { |
|
|
|
ListenableFuture<List<AttributeKvEntry>> ssAttrFuture = attributesService.findAll(edge.getTenantId(), entityId, DataConstants.SERVER_SCOPE); |
|
|
|
Futures.addCallback(ssAttrFuture, new FutureCallback<List<AttributeKvEntry>>() { |
|
|
|
@Override |
|
|
|
public void onSuccess(@Nullable List<AttributeKvEntry> ssAttributes) { |
|
|
|
if (ssAttributes != null && !ssAttributes.isEmpty()) { |
|
|
|
try { |
|
|
|
Map<String, Object> entityData = new HashMap<>(); |
|
|
|
ObjectNode attributes = mapper.createObjectNode(); |
|
|
|
for (AttributeKvEntry attr : ssAttributes) { |
|
|
|
if (attr.getDataType() == DataType.BOOLEAN && attr.getBooleanValue().isPresent()) { |
|
|
|
attributes.put(attr.getKey(), attr.getBooleanValue().get()); |
|
|
|
} else if (attr.getDataType() == DataType.DOUBLE && attr.getDoubleValue().isPresent()) { |
|
|
|
attributes.put(attr.getKey(), attr.getDoubleValue().get()); |
|
|
|
} else if (attr.getDataType() == DataType.LONG && attr.getLongValue().isPresent()) { |
|
|
|
attributes.put(attr.getKey(), attr.getLongValue().get()); |
|
|
|
} else { |
|
|
|
attributes.put(attr.getKey(), attr.getValueAsString()); |
|
|
|
} |
|
|
|
return Futures.transform(ssAttrFuture, ssAttributes -> { |
|
|
|
if (ssAttributes != null && !ssAttributes.isEmpty()) { |
|
|
|
try { |
|
|
|
Map<String, Object> entityData = new HashMap<>(); |
|
|
|
ObjectNode attributes = mapper.createObjectNode(); |
|
|
|
for (AttributeKvEntry attr : ssAttributes) { |
|
|
|
if (attr.getDataType() == DataType.BOOLEAN && attr.getBooleanValue().isPresent()) { |
|
|
|
attributes.put(attr.getKey(), attr.getBooleanValue().get()); |
|
|
|
} else if (attr.getDataType() == DataType.DOUBLE && attr.getDoubleValue().isPresent()) { |
|
|
|
attributes.put(attr.getKey(), attr.getDoubleValue().get()); |
|
|
|
} else if (attr.getDataType() == DataType.LONG && attr.getLongValue().isPresent()) { |
|
|
|
attributes.put(attr.getKey(), attr.getLongValue().get()); |
|
|
|
} else { |
|
|
|
attributes.put(attr.getKey(), attr.getValueAsString()); |
|
|
|
} |
|
|
|
entityData.put("kv", attributes); |
|
|
|
entityData.put("scope", DataConstants.SERVER_SCOPE); |
|
|
|
JsonNode entityBody = mapper.valueToTree(entityData); |
|
|
|
log.debug("Sending attributes data msg, entityId [{}], attributes [{}]", entityId, entityBody); |
|
|
|
saveEdgeEvent(edge.getTenantId(), |
|
|
|
edge.getId(), |
|
|
|
edgeEventType, |
|
|
|
ActionType.ATTRIBUTES_UPDATED, |
|
|
|
entityId, |
|
|
|
entityBody); |
|
|
|
} catch (Exception e) { |
|
|
|
log.error("[{}] Failed to send attribute updates to the edge", edge.getName(), e); |
|
|
|
} |
|
|
|
entityData.put("kv", attributes); |
|
|
|
entityData.put("scope", DataConstants.SERVER_SCOPE); |
|
|
|
JsonNode entityBody = mapper.valueToTree(entityData); |
|
|
|
log.debug("Sending attributes data msg, entityId [{}], attributes [{}]", entityId, entityBody); |
|
|
|
saveEdgeEvent(edge.getTenantId(), |
|
|
|
edge.getId(), |
|
|
|
edgeEventType, |
|
|
|
ActionType.ATTRIBUTES_UPDATED, |
|
|
|
entityId, |
|
|
|
entityBody); |
|
|
|
} catch (Exception e) { |
|
|
|
log.error("[{}] Failed to send attribute updates to the edge", edge.getName(), e); |
|
|
|
throw new RuntimeException("[" + edge.getName() + "] Failed to send attribute updates to the edge", e); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
@Override |
|
|
|
public void onFailure(Throwable t) { |
|
|
|
|
|
|
|
} |
|
|
|
return null; |
|
|
|
}, dbCallbackExecutorService); |
|
|
|
|
|
|
|
// TODO: voba - push shared attributes to edge?
|
|
|
|
ListenableFuture<List<AttributeKvEntry>> shAttrFuture = attributesService.findAll(edge.getTenantId(), entityId, DataConstants.SHARED_SCOPE); |
|
|
|
ListenableFuture<List<AttributeKvEntry>> clAttrFuture = attributesService.findAll(edge.getTenantId(), entityId, DataConstants.CLIENT_SCOPE); |
|
|
|
// ListenableFuture<List<AttributeKvEntry>> shAttrFuture = attributesService.findAll(edge.getTenantId(), entityId, DataConstants.SHARED_SCOPE);
|
|
|
|
// ListenableFuture<List<AttributeKvEntry>> clAttrFuture = attributesService.findAll(edge.getTenantId(), entityId, DataConstants.CLIENT_SCOPE);
|
|
|
|
} else { |
|
|
|
return Futures.immediateFuture(null); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
@ -346,7 +389,7 @@ public class DefaultSyncEdgeService implements SyncEdgeService { |
|
|
|
} |
|
|
|
|
|
|
|
@Override |
|
|
|
public void processRelationRequestMsg(Edge edge, RelationRequestMsg relationRequestMsg) { |
|
|
|
public ListenableFuture<Void> processRelationRequestMsg(Edge edge, RelationRequestMsg relationRequestMsg) { |
|
|
|
EntityId entityId = EntityIdFactory.getByTypeAndUuid( |
|
|
|
EntityType.valueOf(relationRequestMsg.getEntityType()), |
|
|
|
new UUID(relationRequestMsg.getEntityIdMSB(), relationRequestMsg.getEntityIdLSB())); |
|
|
|
@ -355,39 +398,33 @@ public class DefaultSyncEdgeService implements SyncEdgeService { |
|
|
|
futures.add(findRelationByQuery(edge, entityId, EntitySearchDirection.FROM)); |
|
|
|
futures.add(findRelationByQuery(edge, entityId, EntitySearchDirection.TO)); |
|
|
|
ListenableFuture<List<List<EntityRelation>>> relationsListFuture = Futures.allAsList(futures); |
|
|
|
Futures.addCallback(relationsListFuture, new FutureCallback<List<List<EntityRelation>>>() { |
|
|
|
@Override |
|
|
|
public void onSuccess(@Nullable List<List<EntityRelation>> relationsList) { |
|
|
|
try { |
|
|
|
if (!relationsList.isEmpty()) { |
|
|
|
for (List<EntityRelation> entityRelations : relationsList) { |
|
|
|
log.trace("[{}] [{}] [{}] relation(s) are going to be pushed to edge.", edge.getId(), entityId, entityRelations.size()); |
|
|
|
for (EntityRelation relation : entityRelations) { |
|
|
|
try { |
|
|
|
if (!relation.getFrom().getEntityType().equals(EntityType.EDGE) && |
|
|
|
!relation.getTo().getEntityType().equals(EntityType.EDGE)) { |
|
|
|
saveEdgeEvent(edge.getTenantId(), |
|
|
|
edge.getId(), |
|
|
|
EdgeEventType.RELATION, |
|
|
|
ActionType.ADDED, |
|
|
|
null, |
|
|
|
mapper.valueToTree(relation)); |
|
|
|
} |
|
|
|
} catch (Exception e) { |
|
|
|
log.error("Exception during loading relation [{}] to edge on sync!", relation, e); |
|
|
|
return Futures.transform(relationsListFuture, relationsList -> { |
|
|
|
try { |
|
|
|
if (relationsList != null && !relationsList.isEmpty()) { |
|
|
|
for (List<EntityRelation> entityRelations : relationsList) { |
|
|
|
log.trace("[{}] [{}] [{}] relation(s) are going to be pushed to edge.", edge.getId(), entityId, entityRelations.size()); |
|
|
|
for (EntityRelation relation : entityRelations) { |
|
|
|
try { |
|
|
|
if (!relation.getFrom().getEntityType().equals(EntityType.EDGE) && |
|
|
|
!relation.getTo().getEntityType().equals(EntityType.EDGE)) { |
|
|
|
saveEdgeEvent(edge.getTenantId(), |
|
|
|
edge.getId(), |
|
|
|
EdgeEventType.RELATION, |
|
|
|
ActionType.ADDED, |
|
|
|
null, |
|
|
|
mapper.valueToTree(relation)); |
|
|
|
} |
|
|
|
} catch (Exception e) { |
|
|
|
log.error("Exception during loading relation [{}] to edge on sync!", relation, e); |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
} catch (Exception e) { |
|
|
|
log.error("Exception during loading relation(s) to edge on sync!", e); |
|
|
|
} |
|
|
|
} catch (Exception e) { |
|
|
|
log.error("Exception during loading relation(s) to edge on sync!", e); |
|
|
|
throw new RuntimeException("Exception during loading relation(s) to edge on sync!", e); |
|
|
|
} |
|
|
|
|
|
|
|
@Override |
|
|
|
public void onFailure(Throwable t) { |
|
|
|
log.error("Exception during loading relation(s) to edge on sync!", t); |
|
|
|
} |
|
|
|
return null; |
|
|
|
}, dbCallbackExecutorService); |
|
|
|
} |
|
|
|
|
|
|
|
@ -398,22 +435,26 @@ public class DefaultSyncEdgeService implements SyncEdgeService { |
|
|
|
} |
|
|
|
|
|
|
|
@Override |
|
|
|
public void processDeviceCredentialsRequestMsg(Edge edge, DeviceCredentialsRequestMsg deviceCredentialsRequestMsg) { |
|
|
|
public ListenableFuture<Void> processDeviceCredentialsRequestMsg(Edge edge, DeviceCredentialsRequestMsg deviceCredentialsRequestMsg) { |
|
|
|
if (deviceCredentialsRequestMsg.getDeviceIdMSB() != 0 && deviceCredentialsRequestMsg.getDeviceIdLSB() != 0) { |
|
|
|
DeviceId deviceId = new DeviceId(new UUID(deviceCredentialsRequestMsg.getDeviceIdMSB(), deviceCredentialsRequestMsg.getDeviceIdLSB())); |
|
|
|
saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.DEVICE, ActionType.CREDENTIALS_UPDATED, deviceId, null); |
|
|
|
ListenableFuture<EdgeEvent> future = saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.DEVICE, ActionType.CREDENTIALS_UPDATED, deviceId, null); |
|
|
|
return Futures.transform(future, edgeEvent -> null, dbCallbackExecutorService); |
|
|
|
} |
|
|
|
return Futures.immediateFuture(null); |
|
|
|
} |
|
|
|
|
|
|
|
@Override |
|
|
|
public void processUserCredentialsRequestMsg(Edge edge, UserCredentialsRequestMsg userCredentialsRequestMsg) { |
|
|
|
public ListenableFuture<Void> processUserCredentialsRequestMsg(Edge edge, UserCredentialsRequestMsg userCredentialsRequestMsg) { |
|
|
|
if (userCredentialsRequestMsg.getUserIdMSB() != 0 && userCredentialsRequestMsg.getUserIdLSB() != 0) { |
|
|
|
UserId userId = new UserId(new UUID(userCredentialsRequestMsg.getUserIdMSB(), userCredentialsRequestMsg.getUserIdLSB())); |
|
|
|
saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.USER, ActionType.CREDENTIALS_UPDATED, userId, null); |
|
|
|
ListenableFuture<EdgeEvent> future = saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.USER, ActionType.CREDENTIALS_UPDATED, userId, null); |
|
|
|
return Futures.transform(future, edgeEvent -> null, dbCallbackExecutorService); |
|
|
|
} |
|
|
|
return Futures.immediateFuture(null); |
|
|
|
} |
|
|
|
|
|
|
|
private void saveEdgeEvent(TenantId tenantId, |
|
|
|
private ListenableFuture<EdgeEvent> saveEdgeEvent(TenantId tenantId, |
|
|
|
EdgeId edgeId, |
|
|
|
EdgeEventType edgeEventType, |
|
|
|
ActionType edgeEventAction, |
|
|
|
@ -431,6 +472,6 @@ public class DefaultSyncEdgeService implements SyncEdgeService { |
|
|
|
edgeEvent.setEntityId(entityId.getId()); |
|
|
|
} |
|
|
|
edgeEvent.setEntityBody(entityBody); |
|
|
|
edgeEventService.saveAsync(edgeEvent); |
|
|
|
return edgeEventService.saveAsync(edgeEvent); |
|
|
|
} |
|
|
|
} |
|
|
|
|
deaflynx
6 years ago