diff --git a/application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java b/application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java index 1bb30d6960..6d368fbed8 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java @@ -16,6 +16,7 @@ package org.thingsboard.server.service.edge; import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.node.ObjectNode; import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; @@ -23,6 +24,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.common.util.JacksonUtil; import org.thingsboard.common.util.ThingsBoardThreadFactory; import org.thingsboard.server.cluster.TbClusterService; import org.thingsboard.server.common.data.EdgeUtils; @@ -67,6 +69,8 @@ import java.util.concurrent.Executors; @Slf4j public class DefaultEdgeNotificationService implements EdgeNotificationService { + public static final String EDGE_IS_ROOT_BODY_KEY = "isRoot"; + @Autowired private EdgeService edgeService; @@ -142,7 +146,9 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { public Edge setEdgeRootRuleChain(TenantId tenantId, Edge edge, RuleChainId ruleChainId) throws Exception { edge.setRootRuleChainId(ruleChainId); Edge savedEdge = edgeService.saveEdge(edge); - saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.RULE_CHAIN, EdgeEventActionType.UPDATED, ruleChainId, null).get(); + ObjectNode isRootBody = JacksonUtil.OBJECT_MAPPER.createObjectNode(); + isRootBody.put(EDGE_IS_ROOT_BODY_KEY, Boolean.TRUE); + saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.RULE_CHAIN, EdgeEventActionType.UPDATED, ruleChainId, isRootBody).get(); return savedEdge; } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java index 9571319226..28cc32efa3 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java @@ -524,7 +524,7 @@ public final class EdgeGrpcSession implements Closeable { case CUSTOMER: return ctx.getCustomerProcessor().convertCustomerEventToDownlink(edgeEvent); case RULE_CHAIN: - return ctx.getRuleChainProcessor().convertRuleChainEventToDownlink(edge, edgeEvent); + return ctx.getRuleChainProcessor().convertRuleChainEventToDownlink(edgeEvent); case RULE_CHAIN_METADATA: return ctx.getRuleChainProcessor().convertRuleChainMetadataEventToDownlink(edgeEvent, this.edgeVersion); case ALARM: diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/RuleChainMsgConstructor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/RuleChainMsgConstructor.java index 35ac760faa..7f5ae72511 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/RuleChainMsgConstructor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/RuleChainMsgConstructor.java @@ -35,13 +35,13 @@ import org.thingsboard.server.service.edge.rpc.constructor.rule.RuleChainMetadat @TbCoreComponent public class RuleChainMsgConstructor { - public RuleChainUpdateMsg constructRuleChainUpdatedMsg(RuleChainId edgeRootRuleChainId, UpdateMsgType msgType, RuleChain ruleChain) { + public RuleChainUpdateMsg constructRuleChainUpdatedMsg(UpdateMsgType msgType, RuleChain ruleChain, boolean isRoot) { RuleChainUpdateMsg.Builder builder = RuleChainUpdateMsg.newBuilder() .setMsgType(msgType) .setIdMSB(ruleChain.getId().getId().getMostSignificantBits()) .setIdLSB(ruleChain.getId().getId().getLeastSignificantBits()) .setName(ruleChain.getName()) - .setRoot(ruleChain.getId().equals(edgeRootRuleChainId)) + .setRoot(isRoot) .setDebugMode(ruleChain.isDebugMode()) .setConfiguration(JacksonUtil.toString(ruleChain.getConfiguration())); if (ruleChain.getFirstRuleNodeId() != null) { diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/RuleChainEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/RuleChainEdgeProcessor.java index 5704e7d30d..22d8005472 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/RuleChainEdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/RuleChainEdgeProcessor.java @@ -19,7 +19,6 @@ import com.google.common.util.concurrent.ListenableFuture; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; import org.thingsboard.server.common.data.EdgeUtils; -import org.thingsboard.server.common.data.edge.Edge; import org.thingsboard.server.common.data.edge.EdgeEvent; import org.thingsboard.server.common.data.id.RuleChainId; import org.thingsboard.server.common.data.id.TenantId; @@ -33,12 +32,14 @@ import org.thingsboard.server.gen.edge.v1.UpdateMsgType; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.queue.util.TbCoreComponent; +import static org.thingsboard.server.service.edge.DefaultEdgeNotificationService.EDGE_IS_ROOT_BODY_KEY; + @Component @Slf4j @TbCoreComponent public class RuleChainEdgeProcessor extends BaseEdgeProcessor { - public DownlinkMsg convertRuleChainEventToDownlink(Edge edge, EdgeEvent edgeEvent) { + public DownlinkMsg convertRuleChainEventToDownlink(EdgeEvent edgeEvent) { RuleChainId ruleChainId = new RuleChainId(edgeEvent.getEntityId()); DownlinkMsg downlinkMsg = null; switch (edgeEvent.getAction()) { @@ -47,9 +48,15 @@ public class RuleChainEdgeProcessor extends BaseEdgeProcessor { case ASSIGNED_TO_EDGE: RuleChain ruleChain = ruleChainService.findRuleChainById(edgeEvent.getTenantId(), ruleChainId); if (ruleChain != null) { + boolean isRoot = false; + if (edgeEvent.getBody() != null && edgeEvent.getBody().get(EDGE_IS_ROOT_BODY_KEY) != null) { + try { + isRoot = Boolean.parseBoolean(edgeEvent.getBody().get(EDGE_IS_ROOT_BODY_KEY).asText()); + } catch (Exception ignored) {} + } UpdateMsgType msgType = getUpdateMsgType(edgeEvent.getAction()); RuleChainUpdateMsg ruleChainUpdateMsg = - ruleChainMsgConstructor.constructRuleChainUpdatedMsg(edge.getRootRuleChainId(), msgType, ruleChain); + ruleChainMsgConstructor.constructRuleChainUpdatedMsg(msgType, ruleChain, isRoot); downlinkMsg = DownlinkMsg.newBuilder() .setDownlinkMsgId(EdgeUtils.nextPositiveInt()) .addRuleChainUpdateMsg(ruleChainUpdateMsg) diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/TelemetryEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/TelemetryEdgeProcessor.java index f4d713b40f..481ee7762e 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/TelemetryEdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/TelemetryEdgeProcessor.java @@ -48,7 +48,7 @@ import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.EntityViewId; import org.thingsboard.server.common.data.id.RuleChainId; import org.thingsboard.server.common.data.id.TenantId; -import org.thingsboard.server.common.data.kv.AttributeKey; +import org.thingsboard.server.common.data.id.UserId; import org.thingsboard.server.common.data.kv.AttributeKvEntry; import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsgMetaData; @@ -71,9 +71,7 @@ import org.thingsboard.server.queue.util.TbCoreComponent; import javax.annotation.Nullable; import javax.annotation.PostConstruct; import java.util.ArrayList; -import java.util.HashSet; import java.util.List; -import java.util.Set; @Component @Slf4j @@ -325,6 +323,9 @@ public class TelemetryEdgeProcessor extends BaseEdgeProcessor { case CUSTOMER: entityId = new CustomerId(edgeEvent.getEntityId()); break; + case USER: + entityId = new UserId(edgeEvent.getEntityId()); + break; case EDGE: entityId = new EdgeId(edgeEvent.getEntityId()); break; diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/sync/DefaultEdgeRequestsService.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/sync/DefaultEdgeRequestsService.java index 9210d3ceec..4be7ddb1c8 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/sync/DefaultEdgeRequestsService.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/sync/DefaultEdgeRequestsService.java @@ -165,7 +165,8 @@ public class DefaultEdgeRequestsService implements EdgeRequestsService { Map entityData = new HashMap<>(); ObjectNode attributes = JacksonUtil.OBJECT_MAPPER.createObjectNode(); for (AttributeKvEntry attr : ssAttributes) { - if (DefaultDeviceStateService.PERSISTENT_ATTRIBUTES.contains(attr.getKey())) { + if (DefaultDeviceStateService.PERSISTENT_ATTRIBUTES.contains(attr.getKey()) + && !DefaultDeviceStateService.INACTIVITY_TIMEOUT.equals(attr.getKey())) { continue; } if (attr.getDataType() == DataType.BOOLEAN && attr.getBooleanValue().isPresent()) { diff --git a/application/src/test/java/org/thingsboard/server/edge/BaseDeviceEdgeTest.java b/application/src/test/java/org/thingsboard/server/edge/BaseDeviceEdgeTest.java index a3162ebc5f..e5fe165a1e 100644 --- a/application/src/test/java/org/thingsboard/server/edge/BaseDeviceEdgeTest.java +++ b/application/src/test/java/org/thingsboard/server/edge/BaseDeviceEdgeTest.java @@ -319,6 +319,8 @@ abstract public class BaseDeviceEdgeTest extends AbstractEdgeTest { Device device = findDeviceByName("Edge Device 1"); sendAttributesRequestAndVerify(device, DataConstants.SERVER_SCOPE, "{\"key1\":\"value1\"}", "key1", "value1"); + sendAttributesRequestAndVerify(device, DataConstants.SERVER_SCOPE, "{\"inactivityTimeout\":3600000}", + "inactivityTimeout", "3600000"); sendAttributesRequestAndVerify(device, DataConstants.SHARED_SCOPE, "{\"key2\":\"value2\"}", "key2", "value2"); } @@ -576,7 +578,16 @@ abstract public class BaseDeviceEdgeTest extends AbstractEdgeTest { for (TransportProtos.KeyValueProto keyValueProto : attributesUpdatedMsg.getKvList()) { if (keyValueProto.getKey().equals(expectedKey)) { Assert.assertEquals(expectedKey, keyValueProto.getKey()); - Assert.assertEquals(expectedValue, keyValueProto.getStringV()); + switch (keyValueProto.getType()) { + case STRING_V: + Assert.assertEquals(expectedValue, keyValueProto.getStringV()); + break; + case LONG_V: + Assert.assertEquals(Long.parseLong(expectedValue), keyValueProto.getLongV()); + break; + default: + Assert.fail("Unexpected data type: " + keyValueProto.getType()); + } found = true; } } diff --git a/application/src/test/java/org/thingsboard/server/edge/BaseRuleChainEdgeTest.java b/application/src/test/java/org/thingsboard/server/edge/BaseRuleChainEdgeTest.java index 712beeb019..a3539f8053 100644 --- a/application/src/test/java/org/thingsboard/server/edge/BaseRuleChainEdgeTest.java +++ b/application/src/test/java/org/thingsboard/server/edge/BaseRuleChainEdgeTest.java @@ -18,6 +18,7 @@ package org.thingsboard.server.edge; import com.google.protobuf.AbstractMessage; import org.junit.Assert; import org.junit.Test; +import org.thingsboard.server.common.data.edge.Edge; import org.thingsboard.server.common.data.id.RuleChainId; import org.thingsboard.server.common.data.rule.RuleChain; import org.thingsboard.server.common.data.rule.RuleChainMetaData; @@ -165,4 +166,48 @@ abstract public class BaseRuleChainEdgeTest extends AbstractEdgeTest { doPost("/api/ruleChain/metadata", ruleChainMetaData, RuleChainMetaData.class); } + @Test + public void testSetRootRuleChain() throws Exception { + // create rule chain + edgeImitator.expectMessageAmount(1); + RuleChain ruleChain = new RuleChain(); + ruleChain.setName("Edge New Root Rule Chain"); + ruleChain.setType(RuleChainType.EDGE); + RuleChain savedRuleChain = doPost("/api/ruleChain", ruleChain, RuleChain.class); + doPost("/api/edge/" + edge.getUuidId() + + "/ruleChain/" + savedRuleChain.getUuidId(), RuleChain.class); + Assert.assertTrue(edgeImitator.waitForMessages()); + + // set new rule chain as root + RuleChainId currentRootRuleChainId = edge.getRootRuleChainId(); + edgeImitator.expectMessageAmount(1); + doPost("/api/edge/" + edge.getUuidId() + + "/" + savedRuleChain.getUuidId() + "/root", Edge.class); + Assert.assertTrue(edgeImitator.waitForMessages()); + Optional ruleChainUpdateMsgOpt = edgeImitator.findMessageByType(RuleChainUpdateMsg.class); + Assert.assertTrue(ruleChainUpdateMsgOpt.isPresent()); + RuleChainUpdateMsg ruleChainUpdateMsg = ruleChainUpdateMsgOpt.get(); + Assert.assertEquals(UpdateMsgType.ENTITY_UPDATED_RPC_MESSAGE, ruleChainUpdateMsg.getMsgType()); + Assert.assertEquals(ruleChainUpdateMsg.getIdMSB(), savedRuleChain.getUuidId().getMostSignificantBits()); + Assert.assertEquals(ruleChainUpdateMsg.getIdLSB(), savedRuleChain.getUuidId().getLeastSignificantBits()); + Assert.assertTrue(ruleChainUpdateMsg.getRoot()); + + // revert root rule chain + edgeImitator.expectMessageAmount(1); + doPost("/api/edge/" + edge.getUuidId() + + "/" + currentRootRuleChainId.getId() + "/root", Edge.class); + Assert.assertTrue(edgeImitator.waitForMessages()); + + // unassign rule chain from edge + edgeImitator.expectMessageAmount(1); + doDelete("/api/edge/" + edge.getUuidId() + + "/ruleChain/" + savedRuleChain.getUuidId(), RuleChain.class); + Assert.assertTrue(edgeImitator.waitForMessages()); + + // delete rule chain + edgeImitator.expectMessageAmount(1); + doDelete("/api/ruleChain/" + savedRuleChain.getUuidId()) + .andExpect(status().isOk()); + Assert.assertFalse(edgeImitator.waitForMessages(1)); + } } diff --git a/application/src/test/java/org/thingsboard/server/edge/BaseTelemetryEdgeTest.java b/application/src/test/java/org/thingsboard/server/edge/BaseTelemetryEdgeTest.java index 8930166396..4da9f9a001 100644 --- a/application/src/test/java/org/thingsboard/server/edge/BaseTelemetryEdgeTest.java +++ b/application/src/test/java/org/thingsboard/server/edge/BaseTelemetryEdgeTest.java @@ -23,6 +23,7 @@ import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.edge.EdgeEvent; import org.thingsboard.server.common.data.edge.EdgeEventActionType; import org.thingsboard.server.common.data.edge.EdgeEventType; +import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.gen.edge.v1.AttributeDeleteMsg; import org.thingsboard.server.gen.edge.v1.DeviceUpdateMsg; import org.thingsboard.server.gen.edge.v1.EntityDataProto; @@ -67,36 +68,11 @@ abstract public class BaseTelemetryEdgeTest extends AbstractEdgeTest { public void testAttributes() throws Exception { Device device = findDeviceByName("Edge Device 1"); - testAttributesUpdatedMsg(device); + testAttributesUpdatedMsg(device.getId()); testPostAttributesMsg(device); testAttributesDeleteMsg(device); } - private void testAttributesUpdatedMsg(Device device) throws Exception { - String attributesData = "{\"scope\":\"SERVER_SCOPE\",\"kv\":{\"key1\":\"value1\"}}"; - JsonNode attributesEntityData = mapper.readTree(attributesData); - EdgeEvent edgeEvent1 = constructEdgeEvent(tenantId, edge.getId(), EdgeEventActionType.ATTRIBUTES_UPDATED, device.getId().getId(), EdgeEventType.DEVICE, attributesEntityData); - edgeImitator.expectMessageAmount(1); - edgeEventService.saveAsync(edgeEvent1).get(); - clusterService.onEdgeEventUpdate(tenantId, edge.getId()); - Assert.assertTrue(edgeImitator.waitForMessages()); - - AbstractMessage latestMessage = edgeImitator.getLatestMessage(); - Assert.assertTrue(latestMessage instanceof EntityDataProto); - EntityDataProto latestEntityDataMsg = (EntityDataProto) latestMessage; - Assert.assertEquals(device.getUuidId().getMostSignificantBits(), latestEntityDataMsg.getEntityIdMSB()); - Assert.assertEquals(device.getUuidId().getLeastSignificantBits(), latestEntityDataMsg.getEntityIdLSB()); - Assert.assertEquals(device.getId().getEntityType().name(), latestEntityDataMsg.getEntityType()); - Assert.assertEquals("SERVER_SCOPE", latestEntityDataMsg.getPostAttributeScope()); - Assert.assertTrue(latestEntityDataMsg.hasAttributesUpdatedMsg()); - - TransportProtos.PostAttributeMsg attributesUpdatedMsg = latestEntityDataMsg.getAttributesUpdatedMsg(); - Assert.assertEquals(1, attributesUpdatedMsg.getKvCount()); - TransportProtos.KeyValueProto keyValueProto = attributesUpdatedMsg.getKv(0); - Assert.assertEquals("key1", keyValueProto.getKey()); - Assert.assertEquals("value1", keyValueProto.getStringV()); - } - private void testPostAttributesMsg(Device device) throws Exception { String postAttributesData = "{\"scope\":\"SERVER_SCOPE\",\"kv\":{\"key2\":\"value2\"}}"; JsonNode postAttributesEntityData = mapper.readTree(postAttributesData); @@ -226,4 +202,35 @@ abstract public class BaseTelemetryEdgeTest extends AbstractEdgeTest { edgeImitator.setRandomFailuresOnTimeseriesDownlink(false); } + + @Test + public void testAttributesUpdatedMsg_userEntity() throws Exception { + testAttributesUpdatedMsg(tenantAdmin.getId()); + } + + private void testAttributesUpdatedMsg(EntityId entityId) throws Exception { + String attributesData = "{\"scope\":\"SERVER_SCOPE\",\"kv\":{\"key1\":\"value1\"}}"; + JsonNode attributesEntityData = mapper.readTree(attributesData); + EdgeEvent edgeEvent1 = constructEdgeEvent(tenantId, edge.getId(), EdgeEventActionType.ATTRIBUTES_UPDATED, entityId.getId(), EdgeEventType.valueOf(entityId.getEntityType().name()), attributesEntityData); + edgeImitator.expectMessageAmount(1); + edgeEventService.saveAsync(edgeEvent1).get(); + clusterService.onEdgeEventUpdate(tenantId, edge.getId()); + Assert.assertTrue(edgeImitator.waitForMessages()); + + AbstractMessage latestMessage = edgeImitator.getLatestMessage(); + Assert.assertTrue(latestMessage instanceof EntityDataProto); + EntityDataProto latestEntityDataMsg = (EntityDataProto) latestMessage; + Assert.assertEquals(entityId.getId().getMostSignificantBits(), latestEntityDataMsg.getEntityIdMSB()); + Assert.assertEquals(entityId.getId().getLeastSignificantBits(), latestEntityDataMsg.getEntityIdLSB()); + Assert.assertEquals(entityId.getEntityType().name(), latestEntityDataMsg.getEntityType()); + Assert.assertEquals("SERVER_SCOPE", latestEntityDataMsg.getPostAttributeScope()); + Assert.assertTrue(latestEntityDataMsg.hasAttributesUpdatedMsg()); + + TransportProtos.PostAttributeMsg attributesUpdatedMsg = latestEntityDataMsg.getAttributesUpdatedMsg(); + Assert.assertEquals(1, attributesUpdatedMsg.getKvCount()); + TransportProtos.KeyValueProto keyValueProto = attributesUpdatedMsg.getKv(0); + Assert.assertEquals("key1", keyValueProto.getKey()); + Assert.assertEquals("value1", keyValueProto.getStringV()); + } + } diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/edge/AbstractTbMsgPushNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/edge/AbstractTbMsgPushNode.java index f120276df3..73d540ff1d 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/edge/AbstractTbMsgPushNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/edge/AbstractTbMsgPushNode.java @@ -168,6 +168,7 @@ public abstract class AbstractTbMsgPushNode edgePageData = new PageData<>(List.of(edgeId), 1, 1, false); + Mockito.when(edgeService.findRelatedEdgeIdsByEntityId(tenantId, userId, new PageLink(TbMsgPushToEdgeNode.DEFAULT_PAGE_SIZE))).thenReturn(edgePageData); + + TbMsg msg = TbMsg.newMsg(DataConstants.ATTRIBUTES_UPDATED, userId, new TbMsgMetaData(), + TbMsgDataType.JSON, "{}", null, null); + + node.onMsg(ctx, msg); + + verify(edgeEventService).saveAsync(any()); + } }