From 9e4afaa5eeaa1ebc52166341400f7eec533b7f57 Mon Sep 17 00:00:00 2001 From: Andrew Shvayka Date: Tue, 1 May 2018 10:27:22 +0300 Subject: [PATCH] Ability to push msg to different rule chain(s) + msg ack --- .../actors/ruleChain/RuleChainActor.java | 3 + .../RuleChainActorMessageProcessor.java | 33 ++++- .../ruleChain/RuleChainToRuleChainMsg.java | 2 +- .../server/actors/tenant/TenantActor.java | 9 ++ .../AbstractRuleEngineControllerTest.java | 4 + .../rules/RuleEngineNoSqlTestSuite.java | 42 ++++++ .../server/rules/RuleEngineSqlTestSuite.java | 4 +- ...AbstractRuleEngineFlowIntegrationTest.java | 125 ++++++++++++++++++ .../RuleEngineFlowNoSqlIntegrationTest.java | 26 ++++ .../RuleEngineFlowSqlIntegrationTest.java | 3 +- ...leEngineLifecycleNoSqlIntegrationTest.java | 26 ++++ ...RuleEngineLifecycleSqlIntegrationTest.java | 3 +- .../dao/sql/queue/DummySqlMsgQueue.java | 50 ------- .../server/dao/sql/queue/InMemoryMsgKey.java | 29 ++++ .../dao/sql/queue/InMemoryMsgQueue.java | 106 +++++++++++++++ 15 files changed, 403 insertions(+), 62 deletions(-) create mode 100644 application/src/test/java/org/thingsboard/server/rules/RuleEngineNoSqlTestSuite.java create mode 100644 application/src/test/java/org/thingsboard/server/rules/flow/nosql/RuleEngineFlowNoSqlIntegrationTest.java rename application/src/test/java/org/thingsboard/server/rules/flow/{ => sql}/RuleEngineFlowSqlIntegrationTest.java (87%) create mode 100644 application/src/test/java/org/thingsboard/server/rules/lifecycle/nosql/RuleEngineLifecycleNoSqlIntegrationTest.java rename application/src/test/java/org/thingsboard/server/rules/lifecycle/{ => sql}/RuleEngineLifecycleSqlIntegrationTest.java (86%) delete mode 100644 dao/src/main/java/org/thingsboard/server/dao/sql/queue/DummySqlMsgQueue.java create mode 100644 dao/src/main/java/org/thingsboard/server/dao/sql/queue/InMemoryMsgKey.java create mode 100644 dao/src/main/java/org/thingsboard/server/dao/sql/queue/InMemoryMsgQueue.java diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActor.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActor.java index 55525c6e73..4812002ec2 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActor.java +++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActor.java @@ -51,6 +51,9 @@ public class RuleChainActor extends ComponentActor pushMsgToNode(firstNode, msg)); + } else { + pushMsgToNode(firstNode, envelope.getMsg()); + } + } + void onTellNext(RuleNodeToRuleChainTellNextMsg envelope) { checkActive(); RuleNodeId originator = envelope.getOriginator(); @@ -190,8 +199,9 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor pushMsgToNode(targetNodeCtx, queuedMsg)); + enqueueAndForwardMsgCopyToNode(msg, target); break; case RULE_CHAIN: - parent.tell(new RuleChainToRuleChainMsg(new RuleChainId(target.getId()), entityId, msg, true), self); + enqueueAndForwardMsgCopyToChain(msg, target); break; } } //TODO: Ideally this should happen in async way when all targets confirm that the copied messages are successfully written to corresponding target queues. - EntityId ackId = msg.getRuleNodeId() != null ? msg.getRuleNodeId() : msg.getRuleChainId(); queue.ack(msg, ackId.getId(), msg.getClusterPartition()); } } + private void enqueueAndForwardMsgCopyToChain(TbMsg msg, EntityId target) { + RuleChainId targetRCId = new RuleChainId(target.getId()); + TbMsg copyMsg = msg.copy(UUIDs.timeBased(), targetRCId, null, DEFAULT_CLUSTER_PARTITION); + parent.tell(new RuleChainToRuleChainMsg(new RuleChainId(target.getId()), entityId, copyMsg, true), self); + } + + private void enqueueAndForwardMsgCopyToNode(TbMsg msg, EntityId target) { + RuleNodeId targetId = new RuleNodeId(target.getId()); + RuleNodeCtx targetNodeCtx = nodeActors.get(targetId); + TbMsg copy = msg.copy(UUIDs.timeBased(), entityId, targetId, DEFAULT_CLUSTER_PARTITION); + putToQueue(copy, queuedMsg -> pushMsgToNode(targetNodeCtx, queuedMsg)); + } + private void pushToTarget(TbMsg msg, EntityId target) { switch (target.getEntityType()) { case RULE_NODE: diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainToRuleChainMsg.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainToRuleChainMsg.java index f0fca215a8..2b2623bf09 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainToRuleChainMsg.java +++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainToRuleChainMsg.java @@ -26,7 +26,7 @@ import org.thingsboard.server.common.msg.TbMsg; * Created by ashvayka on 19.03.18. */ @Data -final class RuleChainToRuleChainMsg implements TbActorMsg { +public final class RuleChainToRuleChainMsg implements TbActorMsg { private final RuleChainId target; private final RuleChainId source; diff --git a/application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java b/application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java index 0963a6d355..1325b6f230 100644 --- a/application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java +++ b/application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java @@ -22,6 +22,7 @@ import org.thingsboard.server.actors.device.DeviceActor; import org.thingsboard.server.actors.device.DeviceActorToRuleEngineMsg; import org.thingsboard.server.actors.plugin.PluginTerminationMsg; import org.thingsboard.server.actors.ruleChain.RuleChainManagerActor; +import org.thingsboard.server.actors.ruleChain.RuleChainToRuleChainMsg; import org.thingsboard.server.actors.service.ContextBasedCreator; import org.thingsboard.server.actors.service.DefaultActorService; import org.thingsboard.server.actors.shared.plugin.TenantPluginManager; @@ -83,6 +84,9 @@ public class TenantActor extends RuleChainManagerActor { case DEVICE_RPC_REQUEST_TO_DEVICE_ACTOR_MSG: onToDeviceActorMsg((DeviceAwareMsg) msg); break; + case RULE_CHAIN_TO_RULE_CHAIN_MSG: + onRuleChainMsg((RuleChainToRuleChainMsg) msg); + break; default: return false; } @@ -103,6 +107,11 @@ public class TenantActor extends RuleChainManagerActor { ruleChainManager.getRootChainActor().tell(msg, self()); } + private void onRuleChainMsg(RuleChainToRuleChainMsg msg) { + ruleChainManager.getOrCreateActor(context(), msg.getTarget()).tell(msg, self()); + } + + private void onToDeviceActorMsg(DeviceAwareMsg msg) { getOrCreateDeviceActor(msg.getDeviceId()).tell(msg, ActorRef.noSender()); } diff --git a/application/src/test/java/org/thingsboard/server/controller/AbstractRuleEngineControllerTest.java b/application/src/test/java/org/thingsboard/server/controller/AbstractRuleEngineControllerTest.java index 30c7dc3570..5b895b1513 100644 --- a/application/src/test/java/org/thingsboard/server/controller/AbstractRuleEngineControllerTest.java +++ b/application/src/test/java/org/thingsboard/server/controller/AbstractRuleEngineControllerTest.java @@ -27,6 +27,7 @@ import org.thingsboard.server.common.data.page.TimePageData; import org.thingsboard.server.common.data.page.TimePageLink; import org.thingsboard.server.common.data.rule.RuleChain; import org.thingsboard.server.common.data.rule.RuleChainMetaData; +import org.thingsboard.server.dao.queue.MsgQueue; import org.thingsboard.server.dao.rule.RuleChainService; import java.io.IOException; @@ -39,6 +40,9 @@ public class AbstractRuleEngineControllerTest extends AbstractControllerTest { @Autowired protected RuleChainService ruleChainService; + @Autowired + protected MsgQueue msgQueue; + protected RuleChain saveRuleChain(RuleChain ruleChain) throws Exception { return doPost("/api/ruleChain", ruleChain, RuleChain.class); } diff --git a/application/src/test/java/org/thingsboard/server/rules/RuleEngineNoSqlTestSuite.java b/application/src/test/java/org/thingsboard/server/rules/RuleEngineNoSqlTestSuite.java new file mode 100644 index 0000000000..bffe4913ef --- /dev/null +++ b/application/src/test/java/org/thingsboard/server/rules/RuleEngineNoSqlTestSuite.java @@ -0,0 +1,42 @@ +/** + * Copyright © 2016-2018 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.rules; + +import org.cassandraunit.dataset.cql.ClassPathCQLDataSet; +import org.junit.ClassRule; +import org.junit.extensions.cpsuite.ClasspathSuite; +import org.junit.runner.RunWith; +import org.thingsboard.server.dao.CustomCassandraCQLUnit; +import org.thingsboard.server.dao.CustomSqlUnit; + +import java.util.Arrays; + +@RunWith(ClasspathSuite.class) +@ClasspathSuite.ClassnameFilters({ + "org.thingsboard.server.rules.flow.nosql.*Test", + "org.thingsboard.server.rules.lifecycle.nosql.*Test" +}) +public class RuleEngineNoSqlTestSuite { + + @ClassRule + public static CustomCassandraCQLUnit cassandraUnit = + new CustomCassandraCQLUnit( + Arrays.asList( + new ClassPathCQLDataSet("cassandra/schema.cql", false, false), + new ClassPathCQLDataSet("cassandra/system-data.cql", false, false)), + "cassandra-test.yaml", 30000l); + +} diff --git a/application/src/test/java/org/thingsboard/server/rules/RuleEngineSqlTestSuite.java b/application/src/test/java/org/thingsboard/server/rules/RuleEngineSqlTestSuite.java index c4388260e4..7b13e2fd54 100644 --- a/application/src/test/java/org/thingsboard/server/rules/RuleEngineSqlTestSuite.java +++ b/application/src/test/java/org/thingsboard/server/rules/RuleEngineSqlTestSuite.java @@ -24,8 +24,8 @@ import java.util.Arrays; @RunWith(ClasspathSuite.class) @ClasspathSuite.ClassnameFilters({ - "org.thingsboard.server.rules.flow.*Test", - "org.thingsboard.server.rules.lifecycle.*Test"}) + "org.thingsboard.server.rules.flow.sql.*Test", + "org.thingsboard.server.rules.lifecycle.sql.*Test"}) public class RuleEngineSqlTestSuite { @ClassRule diff --git a/application/src/test/java/org/thingsboard/server/rules/flow/AbstractRuleEngineFlowIntegrationTest.java b/application/src/test/java/org/thingsboard/server/rules/flow/AbstractRuleEngineFlowIntegrationTest.java index 2d56772df3..a294816f49 100644 --- a/application/src/test/java/org/thingsboard/server/rules/flow/AbstractRuleEngineFlowIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/rules/flow/AbstractRuleEngineFlowIntegrationTest.java @@ -17,6 +17,7 @@ package org.thingsboard.server.rules.flow; import com.datastax.driver.core.utils.UUIDs; import com.fasterxml.jackson.databind.JsonNode; +import com.google.common.collect.Lists; import lombok.Data; import lombok.extern.slf4j.Slf4j; import org.junit.After; @@ -45,6 +46,7 @@ import org.thingsboard.server.dao.rule.RuleChainService; import java.io.IOException; import java.util.Arrays; import java.util.Collections; +import java.util.List; import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; @@ -186,6 +188,129 @@ public abstract class AbstractRuleEngineFlowIntegrationTest extends AbstractRule Assert.assertEquals("serverAttributeValue1", getMetadata(outEvent).get("ss_serverAttributeKey1").asText()); Assert.assertEquals("serverAttributeValue2", getMetadata(outEvent).get("ss_serverAttributeKey2").asText()); + + List unAckMsgList = Lists.newArrayList(msgQueue.findUnprocessed(ruleChain.getId().getId(), 0L)); + Assert.assertEquals(0, unAckMsgList.size()); + } + + @Test + public void testTwoRuleChainsWithTwoRules() throws Exception { + // Creating Rule Chain + RuleChain rootRuleChain = new RuleChain(); + rootRuleChain.setName("Root Rule Chain"); + rootRuleChain.setTenantId(savedTenant.getId()); + rootRuleChain.setRoot(true); + rootRuleChain.setDebugMode(true); + rootRuleChain = saveRuleChain(rootRuleChain); + Assert.assertNull(rootRuleChain.getFirstRuleNodeId()); + + // Creating Rule Chain + RuleChain secondaryRuleChain = new RuleChain(); + secondaryRuleChain.setName("Secondary Rule Chain"); + secondaryRuleChain.setTenantId(savedTenant.getId()); + secondaryRuleChain.setRoot(false); + secondaryRuleChain.setDebugMode(true); + secondaryRuleChain = saveRuleChain(secondaryRuleChain); + Assert.assertNull(secondaryRuleChain.getFirstRuleNodeId()); + + RuleChainMetaData rootMetaData = new RuleChainMetaData(); + rootMetaData.setRuleChainId(rootRuleChain.getId()); + + RuleNode ruleNode1 = new RuleNode(); + ruleNode1.setName("Simple Rule Node 1"); + ruleNode1.setType(org.thingsboard.rule.engine.metadata.TbGetAttributesNode.class.getName()); + ruleNode1.setDebugMode(true); + TbGetAttributesNodeConfiguration configuration1 = new TbGetAttributesNodeConfiguration(); + configuration1.setServerAttributeNames(Collections.singletonList("serverAttributeKey1")); + ruleNode1.setConfiguration(mapper.valueToTree(configuration1)); + + rootMetaData.setNodes(Collections.singletonList(ruleNode1)); + rootMetaData.setFirstNodeIndex(0); + rootMetaData.addRuleChainConnectionInfo(0, secondaryRuleChain.getId(), "Success", mapper.createObjectNode()); + rootMetaData = saveRuleChainMetaData(rootMetaData); + Assert.assertNotNull(rootMetaData); + + rootRuleChain = getRuleChain(rootRuleChain.getId()); + Assert.assertNotNull(rootRuleChain.getFirstRuleNodeId()); + + + RuleChainMetaData secondaryMetaData = new RuleChainMetaData(); + secondaryMetaData.setRuleChainId(secondaryRuleChain.getId()); + + RuleNode ruleNode2 = new RuleNode(); + ruleNode2.setName("Simple Rule Node 2"); + ruleNode2.setType(org.thingsboard.rule.engine.metadata.TbGetAttributesNode.class.getName()); + ruleNode2.setDebugMode(true); + TbGetAttributesNodeConfiguration configuration2 = new TbGetAttributesNodeConfiguration(); + configuration2.setServerAttributeNames(Collections.singletonList("serverAttributeKey2")); + ruleNode2.setConfiguration(mapper.valueToTree(configuration2)); + + secondaryMetaData.setNodes(Collections.singletonList(ruleNode2)); + secondaryMetaData.setFirstNodeIndex(0); + secondaryMetaData = saveRuleChainMetaData(secondaryMetaData); + Assert.assertNotNull(secondaryMetaData); + + // Saving the device + Device device = new Device(); + device.setName("My device"); + device.setType("default"); + device = doPost("/api/device", device, Device.class); + + attributesService.save(device.getId(), DataConstants.SERVER_SCOPE, + Collections.singletonList(new BaseAttributeKvEntry(new StringDataEntry("serverAttributeKey1", "serverAttributeValue1"), System.currentTimeMillis()))); + attributesService.save(device.getId(), DataConstants.SERVER_SCOPE, + Collections.singletonList(new BaseAttributeKvEntry(new StringDataEntry("serverAttributeKey2", "serverAttributeValue2"), System.currentTimeMillis()))); + + + Thread.sleep(1000); + + // Pushing Message to the system + TbMsg tbMsg = new TbMsg(UUIDs.timeBased(), + "CUSTOM", + device.getId(), + new TbMsgMetaData(), + "{}", null, null, 0L); + actorService.onMsg(new ServiceToRuleEngineMsg(savedTenant.getId(), tbMsg)); + + Thread.sleep(3000); + + TimePageData events = getDebugEvents(savedTenant.getId(), rootRuleChain.getFirstRuleNodeId(), 1000); + + Assert.assertEquals(2, events.getData().size()); + + Event inEvent = events.getData().stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.IN)).findFirst().get(); + Assert.assertEquals(rootRuleChain.getFirstRuleNodeId(), inEvent.getEntityId()); + Assert.assertEquals(device.getId().getId().toString(), inEvent.getBody().get("entityId").asText()); + + Event outEvent = events.getData().stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.OUT)).findFirst().get(); + Assert.assertEquals(rootRuleChain.getFirstRuleNodeId(), outEvent.getEntityId()); + Assert.assertEquals(device.getId().getId().toString(), outEvent.getBody().get("entityId").asText()); + + Assert.assertEquals("serverAttributeValue1", getMetadata(outEvent).get("ss_serverAttributeKey1").asText()); + + RuleChain finalRuleChain = rootRuleChain; + RuleNode lastRuleNode = secondaryMetaData.getNodes().stream().filter(node -> !node.getId().equals(finalRuleChain.getFirstRuleNodeId())).findFirst().get(); + + events = getDebugEvents(savedTenant.getId(), lastRuleNode.getId(), 1000); + + Assert.assertEquals(2, events.getData().size()); + + inEvent = events.getData().stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.IN)).findFirst().get(); + Assert.assertEquals(lastRuleNode.getId(), inEvent.getEntityId()); + Assert.assertEquals(device.getId().getId().toString(), inEvent.getBody().get("entityId").asText()); + + outEvent = events.getData().stream().filter(e -> e.getBody().get("type").asText().equals(DataConstants.OUT)).findFirst().get(); + Assert.assertEquals(lastRuleNode.getId(), outEvent.getEntityId()); + Assert.assertEquals(device.getId().getId().toString(), outEvent.getBody().get("entityId").asText()); + + Assert.assertEquals("serverAttributeValue1", getMetadata(outEvent).get("ss_serverAttributeKey1").asText()); + Assert.assertEquals("serverAttributeValue2", getMetadata(outEvent).get("ss_serverAttributeKey2").asText()); + + List unAckMsgList = Lists.newArrayList(msgQueue.findUnprocessed(rootRuleChain.getId().getId(), 0L)); + Assert.assertEquals(0, unAckMsgList.size()); + + unAckMsgList = Lists.newArrayList(msgQueue.findUnprocessed(secondaryRuleChain.getId().getId(), 0L)); + Assert.assertEquals(0, unAckMsgList.size()); } } diff --git a/application/src/test/java/org/thingsboard/server/rules/flow/nosql/RuleEngineFlowNoSqlIntegrationTest.java b/application/src/test/java/org/thingsboard/server/rules/flow/nosql/RuleEngineFlowNoSqlIntegrationTest.java new file mode 100644 index 0000000000..d073a1de4a --- /dev/null +++ b/application/src/test/java/org/thingsboard/server/rules/flow/nosql/RuleEngineFlowNoSqlIntegrationTest.java @@ -0,0 +1,26 @@ +/** + * Copyright © 2016-2018 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.rules.flow.nosql; + +import org.thingsboard.server.dao.service.DaoNoSqlTest; +import org.thingsboard.server.rules.flow.AbstractRuleEngineFlowIntegrationTest; + +/** + * Created by Valerii Sosliuk on 8/22/2017. + */ +@DaoNoSqlTest +public class RuleEngineFlowNoSqlIntegrationTest extends AbstractRuleEngineFlowIntegrationTest { +} diff --git a/application/src/test/java/org/thingsboard/server/rules/flow/RuleEngineFlowSqlIntegrationTest.java b/application/src/test/java/org/thingsboard/server/rules/flow/sql/RuleEngineFlowSqlIntegrationTest.java similarity index 87% rename from application/src/test/java/org/thingsboard/server/rules/flow/RuleEngineFlowSqlIntegrationTest.java rename to application/src/test/java/org/thingsboard/server/rules/flow/sql/RuleEngineFlowSqlIntegrationTest.java index 18a164ef33..24f288ca59 100644 --- a/application/src/test/java/org/thingsboard/server/rules/flow/RuleEngineFlowSqlIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/rules/flow/sql/RuleEngineFlowSqlIntegrationTest.java @@ -13,10 +13,11 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.thingsboard.server.rules.flow; +package org.thingsboard.server.rules.flow.sql; import org.thingsboard.server.dao.service.DaoSqlTest; import org.thingsboard.server.mqtt.rpc.AbstractMqttServerSideRpcIntegrationTest; +import org.thingsboard.server.rules.flow.AbstractRuleEngineFlowIntegrationTest; /** * Created by Valerii Sosliuk on 8/22/2017. diff --git a/application/src/test/java/org/thingsboard/server/rules/lifecycle/nosql/RuleEngineLifecycleNoSqlIntegrationTest.java b/application/src/test/java/org/thingsboard/server/rules/lifecycle/nosql/RuleEngineLifecycleNoSqlIntegrationTest.java new file mode 100644 index 0000000000..6f66359f62 --- /dev/null +++ b/application/src/test/java/org/thingsboard/server/rules/lifecycle/nosql/RuleEngineLifecycleNoSqlIntegrationTest.java @@ -0,0 +1,26 @@ +/** + * Copyright © 2016-2018 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.rules.lifecycle.nosql; + +import org.thingsboard.server.dao.service.DaoNoSqlTest; +import org.thingsboard.server.rules.lifecycle.AbstractRuleEngineLifecycleIntegrationTest; + +/** + * Created by Valerii Sosliuk on 8/22/2017. + */ +@DaoNoSqlTest +public class RuleEngineLifecycleNoSqlIntegrationTest extends AbstractRuleEngineLifecycleIntegrationTest { +} diff --git a/application/src/test/java/org/thingsboard/server/rules/lifecycle/RuleEngineLifecycleSqlIntegrationTest.java b/application/src/test/java/org/thingsboard/server/rules/lifecycle/sql/RuleEngineLifecycleSqlIntegrationTest.java similarity index 86% rename from application/src/test/java/org/thingsboard/server/rules/lifecycle/RuleEngineLifecycleSqlIntegrationTest.java rename to application/src/test/java/org/thingsboard/server/rules/lifecycle/sql/RuleEngineLifecycleSqlIntegrationTest.java index 004958b2ea..c74c515124 100644 --- a/application/src/test/java/org/thingsboard/server/rules/lifecycle/RuleEngineLifecycleSqlIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/rules/lifecycle/sql/RuleEngineLifecycleSqlIntegrationTest.java @@ -13,10 +13,11 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.thingsboard.server.rules.lifecycle; +package org.thingsboard.server.rules.lifecycle.sql; import org.thingsboard.server.dao.service.DaoSqlTest; import org.thingsboard.server.rules.flow.AbstractRuleEngineFlowIntegrationTest; +import org.thingsboard.server.rules.lifecycle.AbstractRuleEngineLifecycleIntegrationTest; /** * Created by Valerii Sosliuk on 8/22/2017. diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/queue/DummySqlMsgQueue.java b/dao/src/main/java/org/thingsboard/server/dao/sql/queue/DummySqlMsgQueue.java deleted file mode 100644 index 0281d409d2..0000000000 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/queue/DummySqlMsgQueue.java +++ /dev/null @@ -1,50 +0,0 @@ -/** - * Copyright © 2016-2018 The Thingsboard Authors - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.thingsboard.server.dao.sql.queue; - -import com.google.common.util.concurrent.Futures; -import com.google.common.util.concurrent.ListenableFuture; -import lombok.extern.slf4j.Slf4j; -import org.springframework.stereotype.Component; -import org.thingsboard.server.common.msg.TbMsg; -import org.thingsboard.server.dao.queue.MsgQueue; -import org.thingsboard.server.dao.util.SqlDao; - -import java.util.Collections; -import java.util.UUID; - -/** - * Created by ashvayka on 27.04.18. - */ -@Component -@Slf4j -@SqlDao -public class DummySqlMsgQueue implements MsgQueue { - @Override - public ListenableFuture put(TbMsg msg, UUID nodeId, long clusterPartition) { - return Futures.immediateFuture(null); - } - - @Override - public ListenableFuture ack(TbMsg msg, UUID nodeId, long clusterPartition) { - return Futures.immediateFuture(null); - } - - @Override - public Iterable findUnprocessed(UUID nodeId, long clusterPartition) { - return Collections.emptyList(); - } -} diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/queue/InMemoryMsgKey.java b/dao/src/main/java/org/thingsboard/server/dao/sql/queue/InMemoryMsgKey.java new file mode 100644 index 0000000000..2090edf381 --- /dev/null +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/queue/InMemoryMsgKey.java @@ -0,0 +1,29 @@ +/** + * Copyright © 2016-2018 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.dao.sql.queue; + +import lombok.Data; + +import java.util.UUID; + +/** + * Created by ashvayka on 30.04.18. + */ +@Data +public final class InMemoryMsgKey { + final UUID nodeId; + final long clusterPartition; +} diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/queue/InMemoryMsgQueue.java b/dao/src/main/java/org/thingsboard/server/dao/sql/queue/InMemoryMsgQueue.java new file mode 100644 index 0000000000..dd2316869f --- /dev/null +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/queue/InMemoryMsgQueue.java @@ -0,0 +1,106 @@ +/** + * Copyright © 2016-2018 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.dao.sql.queue; + +import com.google.common.util.concurrent.ListenableFuture; +import com.google.common.util.concurrent.ListeningExecutorService; +import com.google.common.util.concurrent.MoreExecutors; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Component; +import org.thingsboard.server.common.msg.TbMsg; +import org.thingsboard.server.dao.queue.MsgQueue; +import org.thingsboard.server.dao.util.SqlDao; + +import javax.annotation.PostConstruct; +import javax.annotation.PreDestroy; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Executors; +import java.util.concurrent.atomic.AtomicLong; + +/** + * Created by ashvayka on 27.04.18. + */ +@Component +@Slf4j +@SqlDao +public class InMemoryMsgQueue implements MsgQueue { + + private ListeningExecutorService queueExecutor; + //TODO: + private AtomicLong pendingMsgCount; + private Map> data = new HashMap<>(); + + @PostConstruct + public void init() { + // Should be always single threaded due to absence of locks. + queueExecutor = MoreExecutors.listeningDecorator(Executors.newSingleThreadExecutor()); + } + + @PreDestroy + public void stop() { + if (queueExecutor == null) { + queueExecutor.shutdownNow(); + } + } + + @Override + public ListenableFuture put(TbMsg msg, UUID nodeId, long clusterPartition) { + return queueExecutor.submit(() -> { + data.computeIfAbsent(new InMemoryMsgKey(nodeId, clusterPartition), key -> new HashMap<>()).put(msg.getId(), msg); + return null; + }); + } + + @Override + public ListenableFuture ack(TbMsg msg, UUID nodeId, long clusterPartition) { + return queueExecutor.submit(() -> { + InMemoryMsgKey key = new InMemoryMsgKey(nodeId, clusterPartition); + Map map = data.get(key); + if (map != null) { + map.remove(msg.getId()); + if (map.isEmpty()) { + data.remove(key); + } + } + return null; + }); + + } + + @Override + public Iterable findUnprocessed(UUID nodeId, long clusterPartition) { + ListenableFuture> list = queueExecutor.submit(() -> { + InMemoryMsgKey key = new InMemoryMsgKey(nodeId, clusterPartition); + Map map = data.get(key); + if (map != null) { + return new ArrayList<>(map.values()); + } else { + return Collections.emptyList(); + } + }); + try { + return list.get(); + } catch (InterruptedException | ExecutionException e) { + throw new RuntimeException(e); + } + } +}