diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/calculated/BaseCalculatedFieldProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/calculated/BaseCalculatedFieldProcessor.java index da3051045f..a7781fe55d 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/calculated/BaseCalculatedFieldProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/calculated/BaseCalculatedFieldProcessor.java @@ -88,10 +88,10 @@ public abstract class BaseCalculatedFieldProcessor extends BaseEdgeProcessor { return Pair.of(isCreated, isNameUpdated); } - protected ListenableFuture pushEventToAllRelatedEdges(TenantId tenantId, EntityId entityId, EdgeEventType type, EdgeEventActionType actionType, EdgeId sourceEdgeId) { + protected ListenableFuture pushEventToAllRelatedEdges(TenantId tenantId, EntityId calculatedFieldOwnerId, EntityId entityId, EdgeEventType type, EdgeEventActionType actionType, EdgeId sourceEdgeId) { List> futures = new ArrayList<>(); PageDataIterableByTenantIdEntityId edgeIds = - new PageDataIterableByTenantIdEntityId<>(edgeCtx.getEdgeService()::findRelatedEdgeIdsByEntityId, tenantId, entityId, RELATED_EDGES_CACHE_ITEMS); + new PageDataIterableByTenantIdEntityId<>(edgeCtx.getEdgeService()::findRelatedEdgeIdsByEntityId, tenantId, calculatedFieldOwnerId, RELATED_EDGES_CACHE_ITEMS); for (EdgeId relatedEdgeId : edgeIds) { if (!relatedEdgeId.equals(sourceEdgeId)) { futures.add(saveEdgeEvent(tenantId, relatedEdgeId, type, actionType, entityId, null)); diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/calculated/CalculatedFieldEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/calculated/CalculatedFieldEdgeProcessor.java index cf28003707..a0212bb4df 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/calculated/CalculatedFieldEdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/calculated/CalculatedFieldEdgeProcessor.java @@ -126,15 +126,15 @@ public class CalculatedFieldEdgeProcessor extends BaseCalculatedFieldProcessor i switch (actionType) { case UPDATED: case ADDED: - EntityId bodyEntityId = JacksonUtil.fromString(edgeNotificationMsg.getBody(), EntityId.class); - if (bodyEntityId != null && - (EntityType.DEVICE.equals(bodyEntityId.getEntityType()) || EntityType.ASSET.equals(bodyEntityId.getEntityType()))) { + EntityId calculatedFieldOwnerId = JacksonUtil.fromString(edgeNotificationMsg.getBody(), EntityId.class); + if (calculatedFieldOwnerId != null && + (EntityType.DEVICE.equals(calculatedFieldOwnerId.getEntityType()) || EntityType.ASSET.equals(calculatedFieldOwnerId.getEntityType()))) { JsonNode body = JacksonUtil.toJsonNode(edgeNotificationMsg.getBody()); EdgeId edgeId = safeGetEdgeId(edgeNotificationMsg.getEdgeIdMSB(), edgeNotificationMsg.getEdgeIdLSB()); return edgeId != null ? saveEdgeEvent(tenantId, edgeId, type, actionType, entityId, body) : - pushEventToAllRelatedEdges(tenantId, entityId, type, actionType, originatorEdgeId); + pushEventToAllRelatedEdges(tenantId, calculatedFieldOwnerId, entityId, type, actionType, originatorEdgeId); } else { return pushEventToAllEdges(tenantId, type, actionType, entityId, originatorEdgeId); } diff --git a/application/src/test/java/org/thingsboard/server/edge/AbstractEdgeTest.java b/application/src/test/java/org/thingsboard/server/edge/AbstractEdgeTest.java index 24fa02f11a..84d879c993 100644 --- a/application/src/test/java/org/thingsboard/server/edge/AbstractEdgeTest.java +++ b/application/src/test/java/org/thingsboard/server/edge/AbstractEdgeTest.java @@ -75,6 +75,7 @@ import org.thingsboard.server.common.data.queue.Queue; 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.security.DeviceCredentials; import org.thingsboard.server.common.data.security.model.JwtSettings; import org.thingsboard.server.controller.AbstractControllerTest; import org.thingsboard.server.dao.edge.EdgeEventService; @@ -565,6 +566,7 @@ abstract public class AbstractEdgeTest extends AbstractControllerTest { protected Device saveDeviceOnCloudAndVerifyDeliveryToEdge() throws Exception { // create device and assign to edge Device savedDevice = saveDevice(StringUtils.randomAlphanumeric(15), thermostatDeviceProfile.getName()); + DeviceCredentials deviceCredentials = doGet("/api/device/" + savedDevice.getId().getId() + "/credentials", DeviceCredentials.class); edgeImitator.expectMessageAmount(3); // device and device profile messages and device credentials doPost("/api/edge/" + edge.getUuidId() + "/device/" + savedDevice.getUuidId(), Device.class); @@ -582,6 +584,15 @@ abstract public class AbstractEdgeTest extends AbstractControllerTest { Assert.assertEquals(UpdateMsgType.ENTITY_CREATED_RPC_MESSAGE, deviceProfileUpdateMsg.getMsgType()); Assert.assertEquals(thermostatDeviceProfile.getUuidId().getMostSignificantBits(), deviceProfileUpdateMsg.getIdMSB()); Assert.assertEquals(thermostatDeviceProfile.getUuidId().getLeastSignificantBits(), deviceProfileUpdateMsg.getIdLSB()); + + Optional deviceCredentialsUpdateMsgOpt = edgeImitator.findMessageByType(DeviceCredentialsUpdateMsg.class); + Assert.assertTrue(deviceCredentialsUpdateMsgOpt.isPresent()); + DeviceCredentialsUpdateMsg deviceCredentialsUpdateMsg = deviceCredentialsUpdateMsgOpt.get(); + DeviceCredentials deviceCredentialsMsg = JacksonUtil.fromString(deviceCredentialsUpdateMsg.getEntity(), DeviceCredentials.class, true); + Assert.assertNotNull(deviceCredentialsMsg); + Assert.assertEquals(savedDevice.getId(), deviceCredentialsMsg.getDeviceId()); + Assert.assertEquals(deviceCredentials, deviceCredentialsMsg); + return savedDevice; } diff --git a/application/src/test/java/org/thingsboard/server/edge/CalculatedFieldEdgeTest.java b/application/src/test/java/org/thingsboard/server/edge/CalculatedFieldEdgeTest.java index da75225d11..5ec48daf63 100644 --- a/application/src/test/java/org/thingsboard/server/edge/CalculatedFieldEdgeTest.java +++ b/application/src/test/java/org/thingsboard/server/edge/CalculatedFieldEdgeTest.java @@ -58,16 +58,15 @@ public class CalculatedFieldEdgeTest extends AbstractEdgeTest { SimpleCalculatedFieldConfiguration config = new SimpleCalculatedFieldConfiguration(); CalculatedField calculatedField = createSimpleCalculatedField(savedDevice.getId(), config); - edgeImitator.expectMessageAmount(SYNC_MESSAGE_COUNT + 4); + edgeImitator.expectMessageAmount(1); CalculatedField savedCalculatedField = doPost("/api/calculatedField", calculatedField, CalculatedField.class); - doPost("/api/edge/sync/" + edge.getId()); Assert.assertTrue(edgeImitator.waitForMessages()); List downlinkMsgs = edgeImitator.getDownlinkMsgs(); AbstractMessage latestMessage = downlinkMsgs.stream().filter(downlinkMsg -> downlinkMsg instanceof CalculatedFieldUpdateMsg).findFirst().get(); Assert.assertTrue(latestMessage instanceof CalculatedFieldUpdateMsg); CalculatedFieldUpdateMsg calculatedFieldUpdateMsg = (CalculatedFieldUpdateMsg) latestMessage; - Assert.assertEquals(UpdateMsgType.ENTITY_UPDATED_RPC_MESSAGE, calculatedFieldUpdateMsg.getMsgType()); + Assert.assertEquals(UpdateMsgType.ENTITY_CREATED_RPC_MESSAGE, calculatedFieldUpdateMsg.getMsgType()); Assert.assertEquals(savedCalculatedField.getUuidId().getMostSignificantBits(), calculatedFieldUpdateMsg.getIdMSB()); Assert.assertEquals(savedCalculatedField.getUuidId().getLeastSignificantBits(), calculatedFieldUpdateMsg.getIdLSB()); CalculatedField calculatedFieldFromMsg = JacksonUtil.fromString(calculatedFieldUpdateMsg.getEntity(), CalculatedField.class, true); @@ -77,11 +76,9 @@ public class CalculatedFieldEdgeTest extends AbstractEdgeTest { Assert.assertEquals(savedDevice.getId(), calculatedFieldFromMsg.getEntityId()); Assert.assertEquals(config, calculatedFieldFromMsg.getConfiguration()); - // update calculatedField - edgeImitator.expectMessageAmount(SYNC_MESSAGE_COUNT + 4); + edgeImitator.expectMessageAmount(1); savedCalculatedField.setName(UPDATED_CF_NAME); savedCalculatedField = doPost("/api/calculatedField", savedCalculatedField, CalculatedField.class); - doPost("/api/edge/sync/" + edge.getId()); Assert.assertTrue(edgeImitator.waitForMessages()); downlinkMsgs = edgeImitator.getDownlinkMsgs(); latestMessage = downlinkMsgs.stream().filter(downlinkMsg -> downlinkMsg instanceof CalculatedFieldUpdateMsg).findFirst().get(); @@ -144,9 +141,8 @@ public class CalculatedFieldEdgeTest extends AbstractEdgeTest { SimpleCalculatedFieldConfiguration config = new SimpleCalculatedFieldConfiguration(); CalculatedField calculatedField = createSimpleCalculatedField(savedDevice.getId(), config); - edgeImitator.expectMessageAmount(SYNC_MESSAGE_COUNT + 4); + edgeImitator.expectMessageAmount(1); CalculatedField savedCalculatedField = doPost("/api/calculatedField", calculatedField, CalculatedField.class); - doPost("/api/edge/sync/" + edge.getId()); Assert.assertTrue(edgeImitator.waitForMessages()); UUID uuid = Uuids.timeBased();