dendi-wang/thingsboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Provide edgeId for EdgeEventSourcingListener to handle cloud entities updated on Edge and process to other Edges

Browse Source
This commit is contained in:
Andrii Landiak 2023-09-05 09:38:48 +03:00
parent 76ea7c95d3
commit e2520ae0f5
24 changed files with 172 additions and 138 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
application/src/main/java/org/thingsboard/server/service/edge/EdgeEventSourcingListener.java
View File

@ -41,7 +41,6 @@ import javax.annotation.PostConstruct;
import static org.thingsboard.server.service.entitiy.DefaultTbNotificationEntityService.edgeTypeByActionType;
/**
* This event listener does not support async event processing because relay on ThreadLocal
* Another possible approach is to implement a special annotation and a bunch of classes similar to TransactionalApplicationListener
@ -71,9 +70,6 @@ public class EdgeEventSourcingListener {
@TransactionalEventListener(fallbackExecution = true)
public void handleEvent(SaveEntityEvent<?> event) {
if (edgeSynchronizationManager.isSync()) {
return;
}
try {
if (!isValidEdgeEventEntity(event.getEntity())) {
return;
@ -81,7 +77,7 @@ public class EdgeEventSourcingListener {
log.trace("[{}] SaveEntityEvent called: {}", event.getTenantId(), event);
EdgeEventActionType action = Boolean.TRUE.equals(event.getAdded()) ? EdgeEventActionType.ADDED : EdgeEventActionType.UPDATED;
tbClusterService.sendNotificationMsgToEdge(event.getTenantId(), null, event.getEntityId(),
null, null, action);
null, null, action, edgeSynchronizationManager.getEdgeId());
} catch (Exception e) {
log.error("[{}] failed to process SaveEntityEvent: {}", event.getTenantId(), event);
}
@ -89,13 +85,11 @@ public class EdgeEventSourcingListener {
@TransactionalEventListener(fallbackExecution = true)
public void handleEvent(DeleteEntityEvent<?> event) {
if (edgeSynchronizationManager.isSync()) {
return;
}
try {
log.trace("[{}] DeleteEntityEvent called: {}", event.getTenantId(), event);
tbClusterService.sendNotificationMsgToEdge(event.getTenantId(), event.getEdgeId(), event.getEntityId(),
JacksonUtil.toString(event.getEntity()), null, EdgeEventActionType.DELETED);
tbClusterService.sendNotificationMsgToEdge(event.getTenantId(), null, event.getEntityId(),
JacksonUtil.toString(event.getEntity()), null, EdgeEventActionType.DELETED,
edgeSynchronizationManager.getEdgeId());
} catch (Exception e) {
log.error("[{}] failed to process DeleteEntityEvent: {}", event.getTenantId(), event);
}
@ -103,13 +97,11 @@ public class EdgeEventSourcingListener {
@TransactionalEventListener(fallbackExecution = true)
public void handleEvent(ActionEntityEvent event) {
if (edgeSynchronizationManager.isSync()) {
return;
}
try {
log.trace("[{}] ActionEntityEvent called: {}", event.getTenantId(), event);
tbClusterService.sendNotificationMsgToEdge(event.getTenantId(), event.getEdgeId(), event.getEntityId(),
event.getBody(), null, edgeTypeByActionType(event.getActionType()));
event.getBody(), null, edgeTypeByActionType(event.getActionType()),
edgeSynchronizationManager.getEdgeId());
} catch (Exception e) {
log.error("[{}] failed to process ActionEntityEvent: {}", event.getTenantId(), event);
}
@ -117,9 +109,6 @@ public class EdgeEventSourcingListener {
@TransactionalEventListener(fallbackExecution = true)
public void handleEvent(RelationActionEvent event) {
if (edgeSynchronizationManager.isSync()) {
return;
}
try {
EntityRelation relation = event.getRelation();
if (relation == null) {
@ -132,7 +121,8 @@ public class EdgeEventSourcingListener {
}
log.trace("[{}] RelationActionEvent called: {}", event.getTenantId(), event);
tbClusterService.sendNotificationMsgToEdge(event.getTenantId(), null, null,
JacksonUtil.toString(relation), EdgeEventType.RELATION, edgeTypeByActionType(event.getActionType()));
JacksonUtil.toString(relation), EdgeEventType.RELATION, edgeTypeByActionType(event.getActionType()),
edgeSynchronizationManager.getEdgeId());
} catch (Exception e) {
log.error("[{}] failed to process RelationActionEvent: {}", event.getTenantId(), event);
}

6
application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java
View File

@ -675,7 +675,7 @@ public final class EdgeGrpcSession implements Closeable {
}
if (uplinkMsg.getDeviceCredentialsUpdateMsgCount() > 0) {
for (DeviceCredentialsUpdateMsg deviceCredentialsUpdateMsg : uplinkMsg.getDeviceCredentialsUpdateMsgList()) {
result.add(ctx.getDeviceProcessor().processDeviceCredentialsMsg(edge.getTenantId(), deviceCredentialsUpdateMsg));
result.add(ctx.getDeviceProcessor().processDeviceCredentialsMsgFromEdge(edge.getTenantId(), edge.getId(), deviceCredentialsUpdateMsg));
}
}
if (uplinkMsg.getAssetProfileUpdateMsgCount() > 0) {
@ -690,7 +690,7 @@ public final class EdgeGrpcSession implements Closeable {
}
if (uplinkMsg.getAlarmUpdateMsgCount() > 0) {
for (AlarmUpdateMsg alarmUpdateMsg : uplinkMsg.getAlarmUpdateMsgList()) {
result.add(ctx.getAlarmProcessor().processAlarmMsg(edge.getTenantId(), alarmUpdateMsg));
result.add(ctx.getAlarmProcessor().processAlarmMsgFromEdge(edge.getTenantId(), edge.getId(), alarmUpdateMsg));
}
}
if (uplinkMsg.getEntityViewUpdateMsgCount() > 0) {
@ -700,7 +700,7 @@ public final class EdgeGrpcSession implements Closeable {
}
if (uplinkMsg.getRelationUpdateMsgCount() > 0) {
for (RelationUpdateMsg relationUpdateMsg : uplinkMsg.getRelationUpdateMsgList()) {
result.add(ctx.getRelationProcessor().processRelationMsg(edge.getTenantId(), relationUpdateMsg));
result.add(ctx.getRelationProcessor().processRelationMsgFromEdge(edge.getTenantId(), edge, relationUpdateMsg));
}
}
if (uplinkMsg.getDashboardUpdateMsgCount() > 0) {

39
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/BaseEdgeProcessor.java
View File

@ -311,7 +311,9 @@ public abstract class BaseEdgeProcessor {
}, dbCallbackExecutorService);
}
protected ListenableFuture<Void> processActionForAllEdges(TenantId tenantId, EdgeEventType type, EdgeEventActionType actionType, EntityId entityId) {
protected ListenableFuture<Void> processActionForAllEdges(TenantId tenantId, EdgeEventType type,
EdgeEventActionType actionType, EntityId entityId,
EdgeId sourceEdgeId) {
List<ListenableFuture<Void>> futures = new ArrayList<>();
if (TenantId.SYS_TENANT_ID.equals(tenantId)) {
PageLink pageLink = new PageLink(DEFAULT_PAGE_SIZE);
@ -319,12 +321,12 @@ public abstract class BaseEdgeProcessor {
do {
tenantsIds = tenantService.findTenantsIds(pageLink);
for (TenantId tenantId1 : tenantsIds.getData()) {
futures.addAll(processActionForAllEdgesByTenantId(tenantId1, type, actionType, entityId, null));
futures.addAll(processActionForAllEdgesByTenantId(tenantId1, type, actionType, entityId, null, sourceEdgeId));
}
pageLink = pageLink.nextPageLink();
} while (tenantsIds.hasNext());
} else {
futures = processActionForAllEdgesByTenantId(tenantId, type, actionType, entityId, null);
futures = processActionForAllEdgesByTenantId(tenantId, type, actionType, entityId, null, sourceEdgeId);
}
return Futures.transform(Futures.allAsList(futures), voids -> null, dbCallbackExecutorService);
}
@ -333,7 +335,8 @@ public abstract class BaseEdgeProcessor {
EdgeEventType type,
EdgeEventActionType actionType,
EntityId entityId,
JsonNode body) {
JsonNode body,
EdgeId sourceEdgeId) {
PageLink pageLink = new PageLink(DEFAULT_PAGE_SIZE);
PageData<Edge> pageData;
List<ListenableFuture<Void>> futures = new ArrayList<>();
@ -341,8 +344,10 @@ public abstract class BaseEdgeProcessor {
pageData = edgeService.findEdgesByTenantId(tenantId, pageLink);
if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) {
for (Edge edge : pageData.getData()) {
if (!edge.getId().equals(sourceEdgeId)) {
futures.add(saveEdgeEvent(tenantId, edge.getId(), type, actionType, entityId, body));
}
}
if (pageData.hasNext()) {
pageLink = pageLink.nextPageLink();
}
@ -385,11 +390,12 @@ public abstract class BaseEdgeProcessor {
EdgeEventType type = EdgeEventType.valueOf(edgeNotificationMsg.getType());
EdgeEventActionType actionType = EdgeEventActionType.valueOf(edgeNotificationMsg.getAction());
EntityId entityId = EntityIdFactory.getByEdgeEventTypeAndUuid(type, new UUID(edgeNotificationMsg.getEntityIdMSB(), edgeNotificationMsg.getEntityIdLSB()));
EdgeId sourceEdgeId = safeGetEdgeId(edgeNotificationMsg.getSourceEdgeIdMSB(), edgeNotificationMsg.getSourceEdgeIdLSB());
if (type.isAllEdgesRelated()) {
return processEntityNotificationForAllEdges(tenantId, type, actionType, entityId);
return processEntityNotificationForAllEdges(tenantId, type, actionType, entityId, sourceEdgeId);
} else {
JsonNode body = JacksonUtil.toJsonNode(edgeNotificationMsg.getBody());
EdgeId edgeId = safeGetEdgeId(edgeNotificationMsg);
EdgeId edgeId = safeGetEdgeId(edgeNotificationMsg.getEdgeIdMSB(), edgeNotificationMsg.getEdgeIdLSB());
switch (actionType) {
case UPDATED:
case CREDENTIALS_UPDATED:
@ -398,18 +404,19 @@ public abstract class BaseEdgeProcessor {
if (edgeId != null) {
return saveEdgeEvent(tenantId, edgeId, type, actionType, entityId, body);
} else {
return processNotificationToRelatedEdges(tenantId, entityId, type, actionType);
return processNotificationToRelatedEdges(tenantId, entityId, type, actionType, sourceEdgeId);
}
case DELETED:
EdgeEventActionType deleted = EdgeEventActionType.DELETED;
if (edgeId != null) {
return saveEdgeEvent(tenantId, edgeId, type, deleted, entityId, body);
} else {
return Futures.transform(Futures.allAsList(processActionForAllEdgesByTenantId(tenantId, type, deleted, entityId, body)),
return Futures.transform(Futures.allAsList(processActionForAllEdgesByTenantId(tenantId, type, deleted, entityId, body, sourceEdgeId)),
voids -> null, dbCallbackExecutorService);
}
case ASSIGNED_TO_EDGE:
case UNASSIGNED_FROM_EDGE:
if (sourceEdgeId == null) {
ListenableFuture<Void> future = saveEdgeEvent(tenantId, edgeId, type, actionType, entityId, body);
return Futures.transformAsync(future, unused -> {
if (type.equals(EdgeEventType.RULE_CHAIN)) {
@ -418,21 +425,23 @@ public abstract class BaseEdgeProcessor {
return Futures.immediateFuture(null);
}
}, dbCallbackExecutorService);
}
default:
return Futures.immediateFuture(null);
}
}
}
private EdgeId safeGetEdgeId(TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) {
if (edgeNotificationMsg.getEdgeIdMSB() != 0 && edgeNotificationMsg.getEdgeIdLSB() != 0) {
return new EdgeId(new UUID(edgeNotificationMsg.getEdgeIdMSB(), edgeNotificationMsg.getEdgeIdLSB()));
protected EdgeId safeGetEdgeId(long edgeIdMSB, long edgeIdLSB) {
if (edgeIdMSB != 0 && edgeIdLSB != 0) {
return new EdgeId(new UUID(edgeIdMSB, edgeIdLSB));
} else {
return null;
}
}
private ListenableFuture<Void> processNotificationToRelatedEdges(TenantId tenantId, EntityId entityId, EdgeEventType type, EdgeEventActionType actionType) {
private ListenableFuture<Void> processNotificationToRelatedEdges(TenantId tenantId, EntityId entityId, EdgeEventType type,
EdgeEventActionType actionType, EdgeId sourceEdgeId) {
PageLink pageLink = new PageLink(DEFAULT_PAGE_SIZE);
PageData<EdgeId> pageData;
List<ListenableFuture<Void>> futures = new ArrayList<>();
@ -440,8 +449,10 @@ public abstract class BaseEdgeProcessor {
pageData = edgeService.findRelatedEdgeIdsByEntityId(tenantId, entityId, pageLink);
if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) {
for (EdgeId relatedEdgeId : pageData.getData()) {
if (!relatedEdgeId.equals(sourceEdgeId)) {
futures.add(saveEdgeEvent(tenantId, relatedEdgeId, type, actionType, entityId, null));
}
}
if (pageData.hasNext()) {
pageLink = pageLink.nextPageLink();
}
@ -483,13 +494,13 @@ public abstract class BaseEdgeProcessor {
return Futures.transform(Futures.allAsList(futures), voids -> null, dbCallbackExecutorService);
}
private ListenableFuture<Void> processEntityNotificationForAllEdges(TenantId tenantId, EdgeEventType type, EdgeEventActionType actionType, EntityId entityId) {
private ListenableFuture<Void> processEntityNotificationForAllEdges(TenantId tenantId, EdgeEventType type, EdgeEventActionType actionType, EntityId entityId, EdgeId sourceEdgeId) {
switch (actionType) {
case ADDED:
case UPDATED:
case DELETED:
case CREDENTIALS_UPDATED: // used by USER entity
return processActionForAllEdges(tenantId, type, actionType, entityId);
return processActionForAllEdges(tenantId, type, actionType, entityId, sourceEdgeId);
default:
return Futures.immediateFuture(null);
}

21
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/alarm/AlarmEdgeProcessor.java
View File

@ -47,6 +47,16 @@ import java.util.UUID;
@TbCoreComponent
public class AlarmEdgeProcessor extends BaseAlarmProcessor {
public ListenableFuture<Void> processAlarmMsgFromEdge(TenantId tenantId, EdgeId edgeId, AlarmUpdateMsg alarmUpdateMsg) {
log.trace("[{}] processAlarmMsgFromEdge [{}]", tenantId, alarmUpdateMsg);
try {
edgeSynchronizationManager.getSync().set(edgeId);
return processAlarmMsg(tenantId, alarmUpdateMsg);
} finally {
edgeSynchronizationManager.getSync().remove();
}
}
public DownlinkMsg convertAlarmEventToDownlink(EdgeEvent edgeEvent) {
AlarmUpdateMsg alarmUpdateMsg =
convertAlarmEventToAlarmMsg(edgeEvent.getTenantId(), edgeEvent.getEntityId(), edgeEvent.getAction(), edgeEvent.getBody());
@ -62,10 +72,12 @@ public class AlarmEdgeProcessor extends BaseAlarmProcessor {
public ListenableFuture<Void> processAlarmNotification(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) throws JsonProcessingException {
EdgeEventActionType actionType = EdgeEventActionType.valueOf(edgeNotificationMsg.getAction());
AlarmId alarmId = new AlarmId(new UUID(edgeNotificationMsg.getEntityIdMSB(), edgeNotificationMsg.getEntityIdLSB()));
EdgeId sourceEdgeId = safeGetEdgeId(edgeNotificationMsg.getSourceEdgeIdMSB(), edgeNotificationMsg.getSourceEdgeIdLSB());
switch (actionType) {
case DELETED:
Alarm deletedAlarm = JacksonUtil.OBJECT_MAPPER.readValue(edgeNotificationMsg.getBody(), Alarm.class);
List<ListenableFuture<Void>> delFutures = pushEventToAllRelatedEdges(tenantId, deletedAlarm.getOriginator(), alarmId, actionType, JacksonUtil.OBJECT_MAPPER.valueToTree(deletedAlarm));
List<ListenableFuture<Void>> delFutures = pushEventToAllRelatedEdges(tenantId, deletedAlarm.getOriginator(),
alarmId, actionType, JacksonUtil.OBJECT_MAPPER.valueToTree(deletedAlarm), sourceEdgeId);
return Futures.transform(Futures.allAsList(delFutures), voids -> null, dbCallbackExecutorService);
default:
ListenableFuture<Alarm> alarmFuture = alarmService.findAlarmByIdAsync(tenantId, alarmId);
@ -77,13 +89,14 @@ public class AlarmEdgeProcessor extends BaseAlarmProcessor {
if (type == null) {
return Futures.immediateFuture(null);
}
List<ListenableFuture<Void>> futures = pushEventToAllRelatedEdges(tenantId, alarm.getOriginator(), alarmId, actionType, null);
List<ListenableFuture<Void>> futures = pushEventToAllRelatedEdges(tenantId, alarm.getOriginator(),
alarmId, actionType, null, sourceEdgeId);
return Futures.transform(Futures.allAsList(futures), voids -> null, dbCallbackExecutorService);
}, dbCallbackExecutorService);
}
}
private List<ListenableFuture<Void>> pushEventToAllRelatedEdges(TenantId tenantId, EntityId originatorId, AlarmId alarmId, EdgeEventActionType actionType, JsonNode body) {
private List<ListenableFuture<Void>> pushEventToAllRelatedEdges(TenantId tenantId, EntityId originatorId, AlarmId alarmId, EdgeEventActionType actionType, JsonNode body, EdgeId sourceEdgeId) {
PageLink pageLink = new PageLink(DEFAULT_PAGE_SIZE);
PageData<EdgeId> pageData;
List<ListenableFuture<Void>> futures = new ArrayList<>();
@ -91,6 +104,7 @@ public class AlarmEdgeProcessor extends BaseAlarmProcessor {
pageData = edgeService.findRelatedEdgeIdsByEntityId(tenantId, originatorId, pageLink);
if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) {
for (EdgeId relatedEdgeId : pageData.getData()) {
if (!relatedEdgeId.equals(sourceEdgeId)) {
futures.add(saveEdgeEvent(tenantId,
relatedEdgeId,
EdgeEventType.ALARM,
@ -98,6 +112,7 @@ public class AlarmEdgeProcessor extends BaseAlarmProcessor {
alarmId,
body));
}
}
if (pageData.hasNext()) {
pageLink = pageLink.nextPageLink();
}

33
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/alarm/BaseAlarmProcessor.java
View File

@ -40,7 +40,6 @@ import java.util.UUID;
public abstract class BaseAlarmProcessor extends BaseEdgeProcessor {
public ListenableFuture<Void> processAlarmMsg(TenantId tenantId, AlarmUpdateMsg alarmUpdateMsg) {
log.trace("[{}] processAlarmMsg [{}]", tenantId, alarmUpdateMsg);
EntityId originatorId = getAlarmOriginator(tenantId, alarmUpdateMsg.getOriginatorName(),
EntityType.valueOf(alarmUpdateMsg.getOriginatorType()));
AlarmId alarmId = new AlarmId(new UUID(alarmUpdateMsg.getIdMSB(), alarmUpdateMsg.getIdLSB()));
@ -49,7 +48,7 @@ public abstract class BaseAlarmProcessor extends BaseEdgeProcessor {
return Futures.immediateFuture(null);
}
try {
edgeSynchronizationManager.getSync().set(true);
switch (alarmUpdateMsg.getMsgType()) {
case ENTITY_CREATED_RPC_MESSAGE:
case ENTITY_UPDATED_RPC_MESSAGE:
@ -100,26 +99,11 @@ public abstract class BaseAlarmProcessor extends BaseEdgeProcessor {
} catch (Exception e) {
log.error("[{}] Failed to process alarm update msg [{}]", tenantId, alarmUpdateMsg, e);
return Futures.immediateFailedFuture(e);
} finally {
edgeSynchronizationManager.getSync().remove();
}
return Futures.immediateFuture(null);
}
private EntityId getAlarmOriginator(TenantId tenantId, String entityName, EntityType entityType) {
switch (entityType) {
case DEVICE:
return deviceService.findDeviceByTenantIdAndName(tenantId, entityName).getId();
case ASSET:
return assetService.findAssetByTenantIdAndName(tenantId, entityName).getId();
case ENTITY_VIEW:
return entityViewService.findEntityViewByTenantIdAndName(tenantId, entityName).getId();
default:
return null;
}
}
public AlarmUpdateMsg convertAlarmEventToAlarmMsg(TenantId tenantId, UUID entityId, EdgeEventActionType actionType, JsonNode body) {
protected AlarmUpdateMsg convertAlarmEventToAlarmMsg(TenantId tenantId, UUID entityId, EdgeEventActionType actionType, JsonNode body) {
AlarmId alarmId = new AlarmId(entityId);
UpdateMsgType msgType = getUpdateMsgType(actionType);
switch (actionType) {
@ -138,4 +122,17 @@ public abstract class BaseAlarmProcessor extends BaseEdgeProcessor {
}
return null;
}
private EntityId getAlarmOriginator(TenantId tenantId, String entityName, EntityType entityType) {
switch (entityType) {
case DEVICE:
return deviceService.findDeviceByTenantIdAndName(tenantId, entityName).getId();
case ASSET:
return assetService.findAssetByTenantIdAndName(tenantId, entityName).getId();
case ENTITY_VIEW:
return entityViewService.findEntityViewByTenantIdAndName(tenantId, entityName).getId();
default:
return null;
}
}
}

2
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/asset/AssetEdgeProcessor.java
View File

@ -57,7 +57,7 @@ public class AssetEdgeProcessor extends BaseAssetProcessor {
log.trace("[{}] executing processAssetMsgFromEdge [{}] from edge [{}]", tenantId, assetUpdateMsg, edge.getName());
AssetId assetId = new AssetId(new UUID(assetUpdateMsg.getIdMSB(), assetUpdateMsg.getIdLSB()));
try {
edgeSynchronizationManager.getSync().set(true);
edgeSynchronizationManager.getSync().set(edge.getId());
switch (assetUpdateMsg.getMsgType()) {
case ENTITY_CREATED_RPC_MESSAGE:

2
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/asset/AssetProfileEdgeProcessor.java
View File

@ -51,7 +51,7 @@ public class AssetProfileEdgeProcessor extends BaseAssetProfileProcessor {
log.trace("[{}] executing processAssetProfileMsgFromEdge [{}] from edge [{}]", tenantId, assetProfileUpdateMsg, edge.getName());
AssetProfileId assetProfileId = new AssetProfileId(new UUID(assetProfileUpdateMsg.getIdMSB(), assetProfileUpdateMsg.getIdLSB()));
try {
edgeSynchronizationManager.getSync().set(true);
edgeSynchronizationManager.getSync().set(edge.getId());
switch (assetProfileUpdateMsg.getMsgType()) {
case ENTITY_CREATED_RPC_MESSAGE:

2
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/dashboard/DashboardEdgeProcessor.java
View File

@ -52,7 +52,7 @@ public class DashboardEdgeProcessor extends BaseDashboardProcessor {
log.trace("[{}] executing processDashboardMsgFromEdge [{}] from edge [{}]", tenantId, dashboardUpdateMsg, edge.getName());
DashboardId dashboardId = new DashboardId(new UUID(dashboardUpdateMsg.getIdMSB(), dashboardUpdateMsg.getIdLSB()));
try {
edgeSynchronizationManager.getSync().set(true);
edgeSynchronizationManager.getSync().set(edge.getId());
switch (dashboardUpdateMsg.getMsgType()) {
case ENTITY_CREATED_RPC_MESSAGE:

12
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/device/BaseDeviceProcessor.java
View File

@ -16,7 +16,6 @@
package org.thingsboard.server.service.edge.rpc.processor.device;
import com.datastax.oss.driver.api.core.uuid.Uuids;
import com.google.common.util.concurrent.ListenableFuture;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.util.Pair;
@ -104,34 +103,27 @@ public abstract class BaseDeviceProcessor extends BaseEdgeProcessor {
return Pair.of(created, deviceNameUpdated);
}
public ListenableFuture<Void> processDeviceCredentialsMsg(TenantId tenantId, DeviceCredentialsUpdateMsg deviceCredentialsUpdateMsg) {
log.debug("[{}] Executing processDeviceCredentialsMsg, deviceCredentialsUpdateMsg [{}]", tenantId, deviceCredentialsUpdateMsg);
protected void updateDeviceCredentials(TenantId tenantId, DeviceCredentialsUpdateMsg deviceCredentialsUpdateMsg) {
DeviceId deviceId = new DeviceId(new UUID(deviceCredentialsUpdateMsg.getDeviceIdMSB(), deviceCredentialsUpdateMsg.getDeviceIdLSB()));
return dbCallbackExecutorService.submit(() -> {
Device device = deviceService.findDeviceById(tenantId, deviceId);
if (device != null) {
log.debug("Updating device credentials for device [{}]. New device credentials Id [{}], value [{}]",
device.getName(), deviceCredentialsUpdateMsg.getCredentialsId(), deviceCredentialsUpdateMsg.getCredentialsValue());
try {
edgeSynchronizationManager.getSync().set(true);
DeviceCredentials deviceCredentials = deviceCredentialsService.findDeviceCredentialsByDeviceId(tenantId, device.getId());
deviceCredentials.setCredentialsType(DeviceCredentialsType.valueOf(deviceCredentialsUpdateMsg.getCredentialsType()));
deviceCredentials.setCredentialsId(deviceCredentialsUpdateMsg.getCredentialsId());
deviceCredentials.setCredentialsValue(deviceCredentialsUpdateMsg.hasCredentialsValue()
? deviceCredentialsUpdateMsg.getCredentialsValue() : null);
deviceCredentialsService.updateDeviceCredentials(tenantId, deviceCredentials);
} catch (Exception e) {
log.error("Can't update device credentials for device [{}], deviceCredentialsUpdateMsg [{}]",
device.getName(), deviceCredentialsUpdateMsg, e);
throw new RuntimeException(e);
} finally {
edgeSynchronizationManager.getSync().remove();
}
} else {
log.warn("Can't find device by id [{}], deviceCredentialsUpdateMsg [{}]", deviceId, deviceCredentialsUpdateMsg);
}
return null;
});
}
}

16
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/device/DeviceEdgeProcessor.java
View File

@ -35,10 +35,12 @@ import org.thingsboard.server.common.data.edge.EdgeEventActionType;
import org.thingsboard.server.common.data.edge.EdgeEventType;
import org.thingsboard.server.common.data.id.CustomerId;
import org.thingsboard.server.common.data.id.DeviceId;
import org.thingsboard.server.common.data.id.EdgeId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.msg.TbMsgType;
import org.thingsboard.server.common.data.rpc.RpcError;
import org.thingsboard.server.common.data.security.DeviceCredentials;
import org.thingsboard.server.common.data.security.DeviceCredentialsType;
import org.thingsboard.server.common.msg.TbMsg;
import org.thingsboard.server.common.msg.TbMsgDataType;
import org.thingsboard.server.common.msg.TbMsgMetaData;
@ -66,7 +68,7 @@ public class DeviceEdgeProcessor extends BaseDeviceProcessor {
log.trace("[{}] executing processDeviceMsgFromEdge [{}] from edge [{}]", tenantId, deviceUpdateMsg, edge.getName());
DeviceId deviceId = new DeviceId(new UUID(deviceUpdateMsg.getIdMSB(), deviceUpdateMsg.getIdLSB()));
try {
edgeSynchronizationManager.getSync().set(true);
edgeSynchronizationManager.getSync().set(edge.getId());
switch (deviceUpdateMsg.getMsgType()) {
case ENTITY_CREATED_RPC_MESSAGE:
@ -95,6 +97,18 @@ public class DeviceEdgeProcessor extends BaseDeviceProcessor {
}
}
public ListenableFuture<Void> processDeviceCredentialsMsgFromEdge(TenantId tenantId, EdgeId edgeId, DeviceCredentialsUpdateMsg deviceCredentialsUpdateMsg) {
log.debug("[{}] Executing processDeviceCredentialsMsgFromEdge, deviceCredentialsUpdateMsg [{}]", tenantId, deviceCredentialsUpdateMsg);
try {
edgeSynchronizationManager.getSync().set(edgeId);
updateDeviceCredentials(tenantId, deviceCredentialsUpdateMsg);
} finally {
edgeSynchronizationManager.getSync().remove();
}
return Futures.immediateFuture(null);
}
private void saveOrUpdateDevice(TenantId tenantId, DeviceId deviceId, DeviceUpdateMsg deviceUpdateMsg, Edge edge) {
CustomerId customerId = safeGetCustomerId(deviceUpdateMsg.getCustomerIdMSB(), deviceUpdateMsg.getCustomerIdLSB());
Pair<Boolean, Boolean> resultPair = super.saveOrUpdateDevice(tenantId, deviceId, deviceUpdateMsg, customerId);

2
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/device/DeviceProfileEdgeProcessor.java
View File

@ -52,7 +52,7 @@ public class DeviceProfileEdgeProcessor extends BaseDeviceProfileProcessor {
log.trace("[{}] executing processDeviceProfileMsgFromEdge [{}] from edge [{}]", tenantId, deviceProfileUpdateMsg, edge.getName());
DeviceProfileId deviceProfileId = new DeviceProfileId(new UUID(deviceProfileUpdateMsg.getIdMSB(), deviceProfileUpdateMsg.getIdLSB()));
try {
edgeSynchronizationManager.getSync().set(true);
edgeSynchronizationManager.getSync().set(edge.getId());
switch (deviceProfileUpdateMsg.getMsgType()) {
case ENTITY_CREATED_RPC_MESSAGE:

2
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/entityview/EntityViewEdgeProcessor.java
View File

@ -55,7 +55,7 @@ public class EntityViewEdgeProcessor extends BaseEntityViewProcessor {
log.trace("[{}] executing processEntityViewMsgFromEdge [{}] from edge [{}]", tenantId, entityViewUpdateMsg, edge.getName());
EntityViewId entityViewId = new EntityViewId(new UUID(entityViewUpdateMsg.getIdMSB(), entityViewUpdateMsg.getIdLSB()));
try {
edgeSynchronizationManager.getSync().set(true);
edgeSynchronizationManager.getSync().set(edge.getId());
switch (entityViewUpdateMsg.getMsgType()) {
case ENTITY_CREATED_RPC_MESSAGE:

4
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/relation/BaseRelationProcessor.java
View File

@ -34,9 +34,7 @@ import java.util.UUID;
public abstract class BaseRelationProcessor extends BaseEdgeProcessor {
public ListenableFuture<Void> processRelationMsg(TenantId tenantId, RelationUpdateMsg relationUpdateMsg) {
log.trace("[{}] processRelationMsg [{}]", tenantId, relationUpdateMsg);
try {
edgeSynchronizationManager.getSync().set(true);
EntityRelation entityRelation = new EntityRelation();
UUID fromUUID = new UUID(relationUpdateMsg.getFromIdMSB(), relationUpdateMsg.getFromIdLSB());
@ -72,8 +70,6 @@ public abstract class BaseRelationProcessor extends BaseEdgeProcessor {
} catch (Exception e) {
log.error("[{}] Failed to process relation update msg [{}]", tenantId, relationUpdateMsg, e);
return Futures.immediateFailedFuture(e);
} finally {
edgeSynchronizationManager.getSync().remove();
}
return Futures.immediateFuture(null);
}

12
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/relation/RelationEdgeProcessor.java
View File

@ -23,6 +23,7 @@ import org.springframework.stereotype.Component;
import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.server.common.data.EdgeUtils;
import org.thingsboard.server.common.data.EntityType;
import org.thingsboard.server.common.data.edge.Edge;
import org.thingsboard.server.common.data.edge.EdgeEvent;
import org.thingsboard.server.common.data.edge.EdgeEventActionType;
import org.thingsboard.server.common.data.edge.EdgeEventType;
@ -45,6 +46,17 @@ import java.util.Set;
@TbCoreComponent
public class RelationEdgeProcessor extends BaseRelationProcessor {
public ListenableFuture<Void> processRelationMsgFromEdge(TenantId tenantId, Edge edge, RelationUpdateMsg relationUpdateMsg) {
log.trace("[{}] executing processRelationMsgFromEdge [{}] from edge [{}]", tenantId, relationUpdateMsg, edge.getName());
try {
edgeSynchronizationManager.getSync().set(edge.getId());
return processRelationMsg(tenantId, relationUpdateMsg);
} finally {
edgeSynchronizationManager.getSync().remove();
}
}
public DownlinkMsg convertRelationEventToDownlink(EdgeEvent edgeEvent) {
EntityRelation entityRelation = JacksonUtil.OBJECT_MAPPER.convertValue(edgeEvent.getBody(), EntityRelation.class);
UpdateMsgType msgType = getUpdateMsgType(edgeEvent.getAction());

6
application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java
View File

@ -475,7 +475,7 @@ public class DefaultTbClusterService implements TbClusterService {
}
@Override
public void sendNotificationMsgToEdge(TenantId tenantId, EdgeId edgeId, EntityId entityId, String body, EdgeEventType type, EdgeEventActionType action) {
public void sendNotificationMsgToEdge(TenantId tenantId, EdgeId edgeId, EntityId entityId, String body, EdgeEventType type, EdgeEventActionType action, EdgeId sourceEdgeId) {
if (!edgesEnabled) {
return;
}
@ -508,6 +508,10 @@ public class DefaultTbClusterService implements TbClusterService {
if (body != null) {
builder.setBody(body);
}
if (sourceEdgeId != null) {
builder.setSourceEdgeIdMSB(sourceEdgeId.getId().getMostSignificantBits());
builder.setSourceEdgeIdLSB(sourceEdgeId.getId().getLeastSignificantBits());
}
TransportProtos.EdgeNotificationMsgProto msg = builder.build();
log.trace("[{}] sending notification to edge service {}", tenantId.getId(), msg);
pushMsgToCore(tenantId, entityId != null ? entityId : tenantId, TransportProtos.ToCoreMsg.newBuilder().setEdgeNotificationMsg(msg).build(), null);

16
application/src/test/java/org/thingsboard/server/controller/AbstractNotifyEntityTest.java
View File

@ -91,7 +91,7 @@ public abstract class AbstractNotifyEntityTest extends AbstractWebTest {
int cntTime = 1;
Mockito.verify(tbClusterService, times(cntTime)).sendNotificationMsgToEdge(Mockito.eq(tenantId),
Mockito.isNull(), Mockito.isNull(), Mockito.any(), Mockito.eq(EdgeEventType.RELATION),
Mockito.eq(edgeTypeByActionType(actionType)));
Mockito.eq(edgeTypeByActionType(actionType)), Mockito.any());
ArgumentMatcher<EntityId> matcherOriginatorId = argument -> argument.equals(relation.getTo());
ArgumentMatcher<HasName> matcherEntityClassEquals = Objects::isNull;
ArgumentMatcher<CustomerId> matcherCustomerId = customerId == null ?
@ -111,7 +111,7 @@ public abstract class AbstractNotifyEntityTest extends AbstractWebTest {
ActionType actionType, int cntTime) {
Mockito.verify(tbClusterService, times(cntTime)).sendNotificationMsgToEdge(Mockito.eq(tenantId),
Mockito.isNull(), Mockito.isNull(), Mockito.any(), Mockito.eq(EdgeEventType.RELATION),
Mockito.eq(edgeTypeByActionType(actionType)));
Mockito.eq(edgeTypeByActionType(actionType)), Mockito.any());
ArgumentMatcher<EntityId> matcherOriginatorId = argument -> argument.getClass().equals(relation.getFrom().getClass());
ArgumentMatcher<HasName> matcherEntityClassEquals = Objects::isNull;
ArgumentMatcher<CustomerId> matcherCustomerId = customerId == null ?
@ -318,13 +318,13 @@ public abstract class AbstractNotifyEntityTest extends AbstractWebTest {
private void testNotificationMsgToEdgeServiceNeverWithActionType(EntityId entityId, ActionType actionType) {
EdgeEventActionType edgeEventActionType = ActionType.CREDENTIALS_UPDATED.equals(actionType) ?
EdgeEventActionType.CREDENTIALS_UPDATED : edgeTypeByActionType(actionType);
Mockito.verify(tbClusterService, never()).sendNotificationMsgToEdge(Mockito.any(),
Mockito.any(), Mockito.any(entityId.getClass()), Mockito.any(), Mockito.any(), Mockito.eq(edgeEventActionType));
Mockito.verify(tbClusterService, never()).sendNotificationMsgToEdge(Mockito.any(), Mockito.any(),
Mockito.any(entityId.getClass()), Mockito.any(), Mockito.any(), Mockito.eq(edgeEventActionType), Mockito.any());
}
private void testNotificationMsgToEdgeServiceNever(EntityId entityId) {
Mockito.verify(tbClusterService, never()).sendNotificationMsgToEdge(Mockito.any(),
Mockito.any(), Mockito.any(entityId.getClass()), Mockito.any(), Mockito.any(), Mockito.any());
Mockito.verify(tbClusterService, never()).sendNotificationMsgToEdge(Mockito.any(), Mockito.any(),
Mockito.any(entityId.getClass()), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any());
}
private void testLogEntityActionNever(EntityId entityId, HasName entity) {
@ -358,13 +358,13 @@ public abstract class AbstractNotifyEntityTest extends AbstractWebTest {
argument -> argument.getClass().equals(entityId.getClass());
Mockito.verify(tbClusterService, times(cntTime)).sendNotificationMsgToEdge(Mockito.eq(tenantId),
Mockito.any(), Mockito.argThat(matcherEntityId), Mockito.any(), Mockito.isNull(),
Mockito.eq(edgeEventActionType));
Mockito.eq(edgeEventActionType), Mockito.any());
}
private void testSendNotificationMsgToEdgeServiceTimeEntityEqAny(TenantId tenantId, ActionType actionType, int cntTime) {
Mockito.verify(tbClusterService, times(cntTime)).sendNotificationMsgToEdge(Mockito.eq(tenantId),
Mockito.any(), Mockito.any(EntityId.class), Mockito.any(), Mockito.isNull(),
Mockito.eq(edgeTypeByActionType(actionType)));
Mockito.eq(edgeTypeByActionType(actionType)), Mockito.any());
}
protected void testBroadcastEntityStateChangeEventTime(EntityId entityId, TenantId tenantId, int cntTime) {

2
application/src/test/java/org/thingsboard/server/edge/AssetEdgeTest.java
View File

@ -213,7 +213,7 @@ public class AssetEdgeTest extends AbstractEdgeTest {
testAutoGeneratedCodeByProtobuf(uplinkMsgBuilder);
edgeImitator.expectResponsesAmount(1);
edgeImitator.expectMessageAmount(1);
edgeImitator.expectMessageAmount(2);
edgeImitator.sendUplinkMsg(uplinkMsgBuilder.build());

8
application/src/test/java/org/thingsboard/server/edge/DeviceEdgeTest.java
View File

@ -548,16 +548,16 @@ public class DeviceEdgeTest extends AbstractEdgeTest {
uplinkMsgBuilder.addDeviceUpdateMsg(deviceUpdateMsgBuilder.build());
edgeImitator.expectResponsesAmount(1);
edgeImitator.expectMessageAmount(1);
edgeImitator.expectMessageAmount(2);
edgeImitator.sendUplinkMsg(uplinkMsgBuilder.build());
Assert.assertTrue(edgeImitator.waitForResponses());
Assert.assertTrue(edgeImitator.waitForMessages());
AbstractMessage latestMessage = edgeImitator.getLatestMessage();
Assert.assertTrue(latestMessage instanceof DeviceCredentialsRequestMsg);
DeviceCredentialsRequestMsg latestDeviceCredentialsRequestMsg = (DeviceCredentialsRequestMsg) latestMessage;
Optional<DeviceCredentialsRequestMsg> deviceCredentialsRequestMsgOpt = edgeImitator.findMessageByType(DeviceCredentialsRequestMsg.class);
Assert.assertTrue(deviceCredentialsRequestMsgOpt.isPresent());
DeviceCredentialsRequestMsg latestDeviceCredentialsRequestMsg = deviceCredentialsRequestMsgOpt.get();
Assert.assertEquals(uuid.getMostSignificantBits(), latestDeviceCredentialsRequestMsg.getDeviceIdMSB());
Assert.assertEquals(uuid.getLeastSignificantBits(), latestDeviceCredentialsRequestMsg.getDeviceIdLSB());

8
application/src/test/java/org/thingsboard/server/service/sync/ie/ExportImportServiceSqlTest.java
View File

@ -555,7 +555,7 @@ public class ExportImportServiceSqlTest extends BaseExportImportServiceTest {
Customer updatedCustomer = importEntity(tenantAdmin2, updatedCustomerEntity).getSavedEntity();
verify(entityActionService).logEntityAction(any(), eq(importedCustomer.getId()), eq(updatedCustomer),
any(), eq(ActionType.UPDATED), isNull());
verify(tbClusterService).sendNotificationMsgToEdge(any(), any(), eq(importedCustomer.getId()), any(), any(), eq(EdgeEventActionType.UPDATED));
verify(tbClusterService).sendNotificationMsgToEdge(any(), any(), eq(importedCustomer.getId()), any(), any(), eq(EdgeEventActionType.UPDATED), any());
Mockito.reset(entityActionService);
@ -572,7 +572,7 @@ public class ExportImportServiceSqlTest extends BaseExportImportServiceTest {
verify(entityActionService).logEntityAction(any(), eq(importedAssetProfile.getId()), eq(importedAssetProfile),
any(), eq(ActionType.ADDED), isNull());
verify(tbClusterService).broadcastEntityStateChangeEvent(any(), eq(importedAssetProfile.getId()), eq(ComponentLifecycleEvent.CREATED));
verify(tbClusterService).sendNotificationMsgToEdge(any(), any(), eq(importedAssetProfile.getId()), any(), any(), eq(EdgeEventActionType.ADDED));
verify(tbClusterService).sendNotificationMsgToEdge(any(), any(), eq(importedAssetProfile.getId()), any(), any(), eq(EdgeEventActionType.ADDED), any());
Asset importedAsset = (Asset) importEntity(tenantAdmin2, getAndClone(entitiesExportData, EntityType.ASSET)).getSavedEntity();
verify(entityActionService).logEntityAction(any(), eq(importedAsset.getId()), eq(importedAsset),
@ -588,14 +588,14 @@ public class ExportImportServiceSqlTest extends BaseExportImportServiceTest {
verify(entityActionService).logEntityAction(any(), eq(importedAsset.getId()), eq(updatedAsset),
any(), eq(ActionType.UPDATED), isNull());
verify(tbClusterService).sendNotificationMsgToEdge(any(), any(), eq(importedAsset.getId()), any(), any(), eq(EdgeEventActionType.UPDATED));
verify(tbClusterService).sendNotificationMsgToEdge(any(), any(), eq(importedAsset.getId()), any(), any(), eq(EdgeEventActionType.UPDATED), any());
DeviceProfile importedDeviceProfile = (DeviceProfile) importEntity(tenantAdmin2, getAndClone(entitiesExportData, EntityType.DEVICE_PROFILE)).getSavedEntity();
verify(entityActionService).logEntityAction(any(), eq(importedDeviceProfile.getId()), eq(importedDeviceProfile),
any(), eq(ActionType.ADDED), isNull());
verify(tbClusterService).onDeviceProfileChange(eq(importedDeviceProfile), any());
verify(tbClusterService).broadcastEntityStateChangeEvent(any(), eq(importedDeviceProfile.getId()), eq(ComponentLifecycleEvent.CREATED));
verify(tbClusterService).sendNotificationMsgToEdge(any(), any(), eq(importedDeviceProfile.getId()), any(), any(), eq(EdgeEventActionType.ADDED));
verify(tbClusterService).sendNotificationMsgToEdge(any(), any(), eq(importedDeviceProfile.getId()), any(), any(), eq(EdgeEventActionType.ADDED), any());
verify(otaPackageStateService).update(eq(importedDeviceProfile), eq(false), eq(false));
Device importedDevice = (Device) importEntity(tenantAdmin2, getAndClone(entitiesExportData, EntityType.DEVICE)).getSavedEntity();

2
common/cluster-api/src/main/java/org/thingsboard/server/cluster/TbClusterService.java
View File

@ -92,6 +92,6 @@ public interface TbClusterService extends TbQueueClusterService {
void pushEdgeSyncResponseToCore(FromEdgeSyncResponse fromEdgeSyncResponse);
void sendNotificationMsgToEdge(TenantId tenantId, EdgeId edgeId, EntityId entityId, String body, EdgeEventType type, EdgeEventActionType action);
void sendNotificationMsgToEdge(TenantId tenantId, EdgeId edgeId, EntityId entityId, String body, EdgeEventType type, EdgeEventActionType action, EdgeId sourceEdgeId);
}

2
common/cluster-api/src/main/proto/queue.proto
View File

@ -744,6 +744,8 @@ message EdgeNotificationMsgProto {
string body = 10;
PostTelemetryMsg postTelemetryMsg = 11;
PostAttributeMsg postAttributesMsg = 12;
int64 sourceEdgeIdMSB = 13;
int64 sourceEdgeIdLSB = 14;
}
/**

6
common/dao-api/src/main/java/org/thingsboard/server/dao/edge/EdgeSynchronizationManager.java
View File

@ -15,9 +15,11 @@
*/
package org.thingsboard.server.dao.edge;
import org.thingsboard.server.common.data.id.EdgeId;
public interface EdgeSynchronizationManager {
ThreadLocal<Boolean> getSync();
ThreadLocal<EdgeId> getSync();
boolean isSync();
EdgeId getEdgeId();
}

8
dao/src/main/java/org/thingsboard/server/dao/edge/DefaultEdgeSynchronizationManager.java
View File

@ -18,17 +18,17 @@ package org.thingsboard.server.dao.edge;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.thingsboard.server.common.data.id.EdgeId;
@Component
@Slf4j
public class DefaultEdgeSynchronizationManager implements EdgeSynchronizationManager {
@Getter
private final ThreadLocal<Boolean> sync = new ThreadLocal<>();
private final ThreadLocal<EdgeId> sync = new ThreadLocal<>();
@Override
public boolean isSync() {
Boolean sync = this.sync.get();
return sync != null && sync;
public EdgeId getEdgeId() {
return this.sync.get();
}
}

1
dao/src/main/java/org/thingsboard/server/dao/eventsourcing/DeleteEntityEvent.java
View File

@ -26,6 +26,5 @@ import org.thingsboard.server.common.data.id.TenantId;
public class DeleteEntityEvent<T> {
private final TenantId tenantId;
private final EntityId entityId;
private final EdgeId edgeId;
private final T entity;
}