From 4c60bd1a4ee614a6767152ac61cd6e14a812728f Mon Sep 17 00:00:00 2001 From: Volodymyr Babak Date: Tue, 13 Dec 2022 16:15:33 +0200 Subject: [PATCH] Edge rule chain service - fixed update of root rule chain in runtime --- .../edge/DefaultEdgeNotificationService.java | 8 +++- .../service/edge/rpc/EdgeGrpcSession.java | 2 +- .../constructor/RuleChainMsgConstructor.java | 4 +- .../rpc/processor/RuleChainEdgeProcessor.java | 13 ++++-- .../server/edge/BaseRuleChainEdgeTest.java | 45 +++++++++++++++++++ 5 files changed, 65 insertions(+), 7 deletions(-) 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/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)); + } }