diff --git a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java index 8b19a6954c..a127b1c910 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java +++ b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java @@ -36,6 +36,7 @@ import org.thingsboard.script.api.js.JsInvokeService; import org.thingsboard.script.api.tbel.TbelInvokeService; import org.thingsboard.server.actors.service.ActorService; import org.thingsboard.server.actors.tenant.DebugTbRateLimits; +import org.thingsboard.server.cache.rule.RuleNodeCache; import org.thingsboard.server.cluster.TbClusterService; import org.thingsboard.server.common.data.event.ErrorEvent; import org.thingsboard.server.common.data.event.LifecycleEvent; @@ -234,6 +235,10 @@ public class ActorSystemContext { @Getter private RuleNodeStateService ruleNodeStateService; + @Autowired + @Getter + private RuleNodeCache ruleNodeCache; + @Autowired private PartitionService partitionService; diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java index 75a037e0af..139d2b1271 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java +++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java @@ -31,6 +31,7 @@ import org.thingsboard.rule.engine.api.RuleEngineAssetProfileCache; import org.thingsboard.rule.engine.api.RuleEngineDeviceProfileCache; import org.thingsboard.rule.engine.api.RuleEngineRpcService; import org.thingsboard.rule.engine.api.RuleEngineTelemetryService; +import org.thingsboard.rule.engine.api.RuleNodeCacheService; import org.thingsboard.rule.engine.api.ScriptEngine; import org.thingsboard.rule.engine.api.SmsService; import org.thingsboard.rule.engine.api.TbContext; @@ -98,6 +99,7 @@ import org.thingsboard.server.dao.widget.WidgetsBundleService; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.queue.TbQueueCallback; import org.thingsboard.server.queue.TbQueueMsgMetadata; +import org.thingsboard.server.service.rule.TbRuleNodeCacheService; import org.thingsboard.server.service.script.RuleNodeJsScriptEngine; import org.thingsboard.server.service.script.RuleNodeTbelScriptEngine; @@ -119,11 +121,13 @@ class DefaultTbContext implements TbContext { private final ActorSystemContext mainCtx; private final String ruleChainName; private final RuleNodeCtx nodeCtx; + private final TbRuleNodeCacheService ruleNodeCacheService; public DefaultTbContext(ActorSystemContext mainCtx, String ruleChainName, RuleNodeCtx nodeCtx) { this.mainCtx = mainCtx; this.ruleChainName = ruleChainName; this.nodeCtx = nodeCtx; + this.ruleNodeCacheService = new DefaultTbRuleNodeCacheService(getSelfId(), mainCtx.getRuleNodeCache()); } @Override @@ -733,6 +737,11 @@ class DefaultTbContext implements TbContext { mainCtx.getRuleNodeStateService().removeByRuleNodeId(getTenantId(), getSelfId()); } + @Override + public RuleNodeCacheService getRuleNodeCacheService() { + return ruleNodeCacheService; + } + @Override public void removeRuleNodeStateForEntity(EntityId entityId) { if (log.isDebugEnabled()) { diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbRuleNodeCacheService.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbRuleNodeCacheService.java new file mode 100644 index 0000000000..9be1663e5d --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbRuleNodeCacheService.java @@ -0,0 +1,90 @@ +/** + * Copyright © 2016-2022 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.actors.ruleChain; + +import lombok.Data; +import lombok.RequiredArgsConstructor; +import org.thingsboard.server.cache.rule.RuleNodeCache; +import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.id.RuleNodeId; +import org.thingsboard.server.common.msg.TbMsg; +import org.thingsboard.server.service.rule.TbRuleNodeCacheService; + +import java.util.List; +import java.util.Set; + +@Data +@RequiredArgsConstructor +public class DefaultTbRuleNodeCacheService implements TbRuleNodeCacheService { + + private final RuleNodeId ruleNodeId; + private final RuleNodeCache cache; + + @Override + public void add(String key, String value) { + cache.add(toRuleNodeCacheKey(key), value); + } + + @Override + public void add(String key, EntityId id) { + cache.add(toRuleNodeCacheKey(key), id); + } + + @Override + public void add(String key, TbMsg value) { + cache.add(toRuleNodeCacheKey(key), value); + } + + @Override + public void removeTbMsgList(String key, List values) { + cache.removeTbMsgList(toRuleNodeCacheKey(key), values); + } + + @Override + public void removeStringList(String key, List values) { + cache.removeStringList(toRuleNodeCacheKey(key), values); + } + + @Override + public void removeEntityIdList(String key, List values) { + cache.removeEntityIdList(toRuleNodeCacheKey(key), values); + } + + @Override + public Set getStrings(String key) { + return cache.getStringSetByKey(toRuleNodeCacheKey(key)); + } + + @Override + public Set getEntityIds(String key) { + return cache.getEntityIdSetByKey(toRuleNodeCacheKey(key)); + } + + @Override + public Set getTbMsgs(String key, String queueName) { + return cache.getTbMsgSetByKey(toRuleNodeCacheKey(key), queueName); + } + + @Override + public void evict(String key) { + cache.evict(key); + } + + private String toRuleNodeCacheKey(String key) { + return String.format("%s::%s", ruleNodeId.toString(), key); + } + +} diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleNodeActorMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleNodeActorMessageProcessor.java index 4d102337aa..4dd00b75f4 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleNodeActorMessageProcessor.java +++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleNodeActorMessageProcessor.java @@ -88,7 +88,7 @@ public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor queue.getName().equals("Main")).findFirst().orElseThrow(() -> { + queues.stream().filter(queue -> queue.getName().equals(DataConstants.MAIN_QUEUE_NAME)).findFirst().orElseThrow(() -> { log.error("Main queue is not configured in thingsboard.yml"); return new RuntimeException("No \"Main\" queue configured!"); }); diff --git a/application/src/main/java/org/thingsboard/server/service/install/update/DefaultDataUpdateService.java b/application/src/main/java/org/thingsboard/server/service/install/update/DefaultDataUpdateService.java index ee34bd7f23..96d5aa3461 100644 --- a/application/src/main/java/org/thingsboard/server/service/install/update/DefaultDataUpdateService.java +++ b/application/src/main/java/org/thingsboard/server/service/install/update/DefaultDataUpdateService.java @@ -30,6 +30,7 @@ import org.thingsboard.rule.engine.flow.TbRuleChainInputNode; import org.thingsboard.rule.engine.flow.TbRuleChainInputNodeConfiguration; import org.thingsboard.rule.engine.profile.TbDeviceProfileNode; import org.thingsboard.rule.engine.profile.TbDeviceProfileNodeConfiguration; +import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.EntityView; import org.thingsboard.server.common.data.Tenant; import org.thingsboard.server.common.data.TenantProfile; @@ -651,8 +652,8 @@ public class DefaultDataUpdateService implements DataUpdateService { private TenantProfileQueueConfiguration getMainQueueConfiguration() { TenantProfileQueueConfiguration mainQueueConfiguration = new TenantProfileQueueConfiguration(); - mainQueueConfiguration.setName("Main"); - mainQueueConfiguration.setTopic("tb_rule_engine.main"); + mainQueueConfiguration.setName(DataConstants.MAIN_QUEUE_NAME); + mainQueueConfiguration.setTopic(DataConstants.MAIN_QUEUE_TOPIC); mainQueueConfiguration.setPollInterval(25); mainQueueConfiguration.setPartitions(10); mainQueueConfiguration.setConsumerPerPartition(true); diff --git a/application/src/main/java/org/thingsboard/server/service/rule/TbRuleNodeCacheService.java b/application/src/main/java/org/thingsboard/server/service/rule/TbRuleNodeCacheService.java new file mode 100644 index 0000000000..e7ac74e024 --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/rule/TbRuleNodeCacheService.java @@ -0,0 +1,22 @@ +/** + * Copyright © 2016-2022 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.service.rule; + +import org.thingsboard.rule.engine.api.RuleNodeCacheService; + +public interface TbRuleNodeCacheService extends RuleNodeCacheService { + +} diff --git a/application/src/test/java/org/thingsboard/server/controller/BaseTenantControllerTest.java b/application/src/test/java/org/thingsboard/server/controller/BaseTenantControllerTest.java index 4cba7c1ee0..e781c1f71a 100644 --- a/application/src/test/java/org/thingsboard/server/controller/BaseTenantControllerTest.java +++ b/application/src/test/java/org/thingsboard/server/controller/BaseTenantControllerTest.java @@ -30,6 +30,7 @@ import org.mockito.Mockito; import org.springframework.test.context.TestPropertySource; import org.springframework.test.web.servlet.ResultActions; import org.thingsboard.common.util.ThingsBoardExecutors; +import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.StringUtils; import org.thingsboard.server.common.data.Tenant; import org.thingsboard.server.common.data.TenantInfo; @@ -427,7 +428,7 @@ public abstract class BaseTenantControllerTest extends AbstractControllerTest { tenantProfileData.setConfiguration(new DefaultTenantProfileConfiguration()); tenantProfile.setProfileData(tenantProfileData); tenantProfile.setIsolatedTbRuleEngine(true); - addQueueConfig(tenantProfile, "Main"); + addQueueConfig(tenantProfile, DataConstants.MAIN_QUEUE_NAME); addQueueConfig(tenantProfile, "Test"); tenantProfile = doPost("/api/tenantProfile", tenantProfile, TenantProfile.class); @@ -456,7 +457,7 @@ public abstract class BaseTenantControllerTest extends AbstractControllerTest { tenantProfileData2.setConfiguration(new DefaultTenantProfileConfiguration()); tenantProfile2.setProfileData(tenantProfileData2); tenantProfile2.setIsolatedTbRuleEngine(true); - addQueueConfig(tenantProfile2, "Main"); + addQueueConfig(tenantProfile2, DataConstants.MAIN_QUEUE_NAME); addQueueConfig(tenantProfile2, "Test"); addQueueConfig(tenantProfile2, "Test2"); tenantProfile2 = doPost("/api/tenantProfile", tenantProfile2, TenantProfile.class); diff --git a/application/src/test/java/org/thingsboard/server/controller/BaseTenantProfileControllerTest.java b/application/src/test/java/org/thingsboard/server/controller/BaseTenantProfileControllerTest.java index 0a4f7ac60b..3c79cd0101 100644 --- a/application/src/test/java/org/thingsboard/server/controller/BaseTenantProfileControllerTest.java +++ b/application/src/test/java/org/thingsboard/server/controller/BaseTenantProfileControllerTest.java @@ -20,6 +20,7 @@ import org.junit.Assert; import org.junit.Test; import org.mockito.ArgumentMatcher; import org.mockito.Mockito; +import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.EntityInfo; import org.thingsboard.server.common.data.StringUtils; import org.thingsboard.server.common.data.Tenant; @@ -341,8 +342,8 @@ public abstract class BaseTenantProfileControllerTest extends AbstractController private void addMainQueueConfig(TenantProfile tenantProfile) { TenantProfileQueueConfiguration mainQueueConfiguration = new TenantProfileQueueConfiguration(); - mainQueueConfiguration.setName("Main"); - mainQueueConfiguration.setTopic("tb_rule_engine.main"); + mainQueueConfiguration.setName(DataConstants.MAIN_QUEUE_NAME); + mainQueueConfiguration.setTopic(DataConstants.MAIN_QUEUE_TOPIC); mainQueueConfiguration.setPollInterval(25); mainQueueConfiguration.setPartitions(10); mainQueueConfiguration.setConsumerPerPartition(true); 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 02406da08a..2bdcc0dc66 100644 --- a/application/src/test/java/org/thingsboard/server/edge/AbstractEdgeTest.java +++ b/application/src/test/java/org/thingsboard/server/edge/AbstractEdgeTest.java @@ -28,6 +28,7 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.test.context.TestPropertySource; import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.server.cluster.TbClusterService; +import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.DeviceProfile; import org.thingsboard.server.common.data.OtaPackageInfo; @@ -427,8 +428,8 @@ abstract public class AbstractEdgeTest extends AbstractControllerTest { UUID queueUUID = new UUID(queueUpdateMsg.getIdMSB(), queueUpdateMsg.getIdLSB()); Queue queue = doGet("/api/queues/" + queueUUID, Queue.class); Assert.assertNotNull(queue); - Assert.assertEquals("Main", queueUpdateMsg.getName()); - Assert.assertEquals("tb_rule_engine.main", queueUpdateMsg.getTopic()); + Assert.assertEquals(DataConstants.MAIN_QUEUE_NAME, queueUpdateMsg.getName()); + Assert.assertEquals(DataConstants.MAIN_QUEUE_TOPIC, queueUpdateMsg.getTopic()); Assert.assertEquals(10, queueUpdateMsg.getPartitions()); Assert.assertEquals(25, queueUpdateMsg.getPollInterval()); testAutoGeneratedCodeByProtobuf(queueUpdateMsg); diff --git a/application/src/test/java/org/thingsboard/server/service/queue/DefaultTbClusterServiceTest.java b/application/src/test/java/org/thingsboard/server/service/queue/DefaultTbClusterServiceTest.java index 27eeda0b5b..766e9c7f9a 100644 --- a/application/src/test/java/org/thingsboard/server/service/queue/DefaultTbClusterServiceTest.java +++ b/application/src/test/java/org/thingsboard/server/service/queue/DefaultTbClusterServiceTest.java @@ -24,6 +24,7 @@ import org.springframework.boot.test.mock.mockito.SpyBean; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringRunner; import org.thingsboard.server.cluster.TbClusterService; +import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.id.QueueId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.queue.Queue; @@ -239,7 +240,7 @@ public class DefaultTbClusterServiceTest { TenantId tenantId = TenantId.SYS_TENANT_ID; Queue queue = new Queue(new QueueId(UUID.randomUUID())); queue.setTenantId(tenantId); - queue.setName("Main"); + queue.setName(DataConstants.MAIN_QUEUE_NAME); queue.setTopic("main"); queue.setPartitions(10); return queue; diff --git a/application/src/test/java/org/thingsboard/server/service/queue/TbMsgPackProcessingContextTest.java b/application/src/test/java/org/thingsboard/server/service/queue/TbMsgPackProcessingContextTest.java index 7a827f0e94..5a666377c4 100644 --- a/application/src/test/java/org/thingsboard/server/service/queue/TbMsgPackProcessingContextTest.java +++ b/application/src/test/java/org/thingsboard/server/service/queue/TbMsgPackProcessingContextTest.java @@ -22,6 +22,7 @@ import org.junit.Test; import org.junit.runner.RunWith; import org.mockito.junit.MockitoJUnitRunner; import org.thingsboard.common.util.ThingsBoardThreadFactory; +import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.queue.common.TbProtoQueueMsg; import org.thingsboard.server.service.queue.processing.TbRuleEngineSubmitStrategy; @@ -69,7 +70,7 @@ public class TbMsgPackProcessingContextTest { TbRuleEngineSubmitStrategy strategyMock = mock(TbRuleEngineSubmitStrategy.class); when(strategyMock.getPendingMap()).thenReturn(messages); - TbMsgPackProcessingContext context = new TbMsgPackProcessingContext("Main", strategyMock, false); + TbMsgPackProcessingContext context = new TbMsgPackProcessingContext(DataConstants.MAIN_QUEUE_NAME, strategyMock, false); for (UUID uuid : messages.keySet()) { final CountDownLatch readyLatch = new CountDownLatch(parallelCount); final CountDownLatch startLatch = new CountDownLatch(1); diff --git a/common/cache/pom.xml b/common/cache/pom.xml index 8c7b3231c2..d75e744180 100644 --- a/common/cache/pom.xml +++ b/common/cache/pom.xml @@ -95,6 +95,10 @@ awaitility test + + org.thingsboard.common + message + diff --git a/common/cache/src/main/java/org/thingsboard/server/cache/rule/DefaultRuleNodeCache.java b/common/cache/src/main/java/org/thingsboard/server/cache/rule/DefaultRuleNodeCache.java new file mode 100644 index 0000000000..8cbdc4e415 --- /dev/null +++ b/common/cache/src/main/java/org/thingsboard/server/cache/rule/DefaultRuleNodeCache.java @@ -0,0 +1,74 @@ +/** + * Copyright © 2016-2022 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.cache.rule; + +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.stereotype.Service; +import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.msg.TbMsg; + +import java.util.Collections; +import java.util.List; +import java.util.Set; + +@Service +@ConditionalOnProperty(prefix = "cache", value = "type", havingValue = "caffeine", matchIfMissing = true) +public class DefaultRuleNodeCache implements RuleNodeCache { + + @Override + public void add(String key, String value) { + } + + @Override + public void add(String key, EntityId value) { + } + + @Override + public void add(String key, TbMsg value) { + } + + @Override + public void removeStringList(String key, List values) { + } + + @Override + public void removeEntityIdList(String key, List values) { + } + + @Override + public void removeTbMsgList(String key, List values) { + } + + @Override + public Set getStringSetByKey(String key) { + return Collections.emptySet(); + } + + @Override + public Set getEntityIdSetByKey(String key) { + return Collections.emptySet(); + } + + @Override + public Set getTbMsgSetByKey(String key, String queueName) { + return Collections.emptySet(); + } + + @Override + public void evict(String key) { + } + +} diff --git a/common/cache/src/main/java/org/thingsboard/server/cache/rule/RedisRuleNodeCache.java b/common/cache/src/main/java/org/thingsboard/server/cache/rule/RedisRuleNodeCache.java new file mode 100644 index 0000000000..49b7ff67bb --- /dev/null +++ b/common/cache/src/main/java/org/thingsboard/server/cache/rule/RedisRuleNodeCache.java @@ -0,0 +1,153 @@ +/** + * Copyright © 2016-2022 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.cache.rule; + +import lombok.RequiredArgsConstructor; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.data.redis.connection.RedisConnection; +import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.stereotype.Service; +import org.springframework.util.SerializationUtils; +import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.msg.TbMsg; +import org.thingsboard.server.common.msg.queue.TbMsgCallback; + +import java.util.Collections; +import java.util.List; +import java.util.Set; +import java.util.stream.Collectors; + +@Service +@ConditionalOnProperty(prefix = "cache", value = "type", havingValue = "redis") +@RequiredArgsConstructor +public class RedisRuleNodeCache implements RuleNodeCache { + + private final RedisConnectionFactory redisConnectionFactory; + + @Override + public void add(String key, String value) { + processAdd(key, value.getBytes()); + } + + @Override + public void add(String key, EntityId value) { + processAdd(key, SerializationUtils.serialize(value)); + } + + @Override + public void add(String key, TbMsg value) { + processAdd(key, TbMsg.toByteArray(value)); + } + + @Override + public void removeStringList(String key, List values) { + processRemove(key, stringListToBytes(values)); + } + + @Override + public void removeEntityIdList(String key, List values) { + processRemove(key, entityIdListToBytes(values)); + } + + @Override + public void removeTbMsgList(String key, List values) { + processRemove(key, tbMsgListToBytes(values)); + } + + @Override + public Set getStringSetByKey(String key) { + return toStringSet(processGetMembers(key)); + } + + @Override + public Set getEntityIdSetByKey(String key) { + return toEntityIdSet(processGetMembers(key)); + } + + @Override + public Set getTbMsgSetByKey(String key, String queueName) { + return toTbMsgSet(processGetMembers(key), queueName); + } + + @Override + public void evict(String key) { + try (RedisConnection connection = redisConnectionFactory.getConnection()) { + connection.del(key.getBytes()); + } + } + + private void processAdd(String key, byte[] value) { + try (RedisConnection connection = redisConnectionFactory.getConnection()) { + connection.setCommands().sAdd(key.getBytes(), value); + } + } + + private void processRemove(String key, byte[][] values) { + if (values.length == 0) { + return; + } + try (RedisConnection connection = redisConnectionFactory.getConnection()) { + connection.setCommands().sRem(key.getBytes(), values); + } + } + + private Set processGetMembers(String key) { + try (RedisConnection connection = redisConnectionFactory.getConnection()) { + Set bytes = connection.setCommands().sMembers(key.getBytes()); + if (bytes == null) { + return Collections.emptySet(); + } + return bytes; + } + } + + private byte[][] stringListToBytes(List values) { + return values.stream() + .map(String::getBytes) + .toArray(byte[][]::new); + } + + private byte[][] entityIdListToBytes(List values) { + return values.stream() + .map(SerializationUtils::serialize) + .toArray(byte[][]::new); + } + + private byte[][] tbMsgListToBytes(List values) { + return values.stream() + .map(TbMsg::toByteArray) + .toArray(byte[][]::new); + } + + private Set toStringSet(Set values) { + return values.stream() + .map(String::new) + .collect(Collectors.toSet()); + } + + private Set toEntityIdSet(Set values) { + return values.stream() + .map(bytes -> (EntityId) SerializationUtils.deserialize(bytes)) + .collect(Collectors.toSet()); + } + + private Set toTbMsgSet(Set values, String queueName) { + return values.stream() + .map(bytes -> TbMsg.fromBytes(queueName, bytes, TbMsgCallback.EMPTY)) + .collect(Collectors.toSet()); + } + +} diff --git a/common/cache/src/main/java/org/thingsboard/server/cache/rule/RuleNodeCache.java b/common/cache/src/main/java/org/thingsboard/server/cache/rule/RuleNodeCache.java new file mode 100644 index 0000000000..3f6697e219 --- /dev/null +++ b/common/cache/src/main/java/org/thingsboard/server/cache/rule/RuleNodeCache.java @@ -0,0 +1,45 @@ +/** + * Copyright © 2016-2022 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.cache.rule; + +import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.msg.TbMsg; + +import java.util.List; +import java.util.Set; + +public interface RuleNodeCache { + + void add(String key, String value); + + void add(String key, EntityId value); + + void add(String key, TbMsg value); + + void removeStringList(String key, List values); + + void removeEntityIdList(String key, List values); + + void removeTbMsgList(String key, List values); + + Set getStringSetByKey(String key); + + Set getEntityIdSetByKey(String key); + + Set getTbMsgSetByKey(String key, String queueName); + + void evict(String key); +} diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java b/common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java index 970fd6177e..ac86c94e9c 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java @@ -118,4 +118,12 @@ public class DataConstants { public static final String MSG_SOURCE_KEY = "source"; public static final String LAST_CONNECTED_GATEWAY = "lastConnectedGateway"; + + public static final String MAIN_QUEUE_NAME = "Main"; + public static final String MAIN_QUEUE_TOPIC = "tb_rule_engine.main"; + public static final String HP_QUEUE_NAME = "HighPriority"; + public static final String HP_QUEUE_TOPIC = "tb_rule_engine.hp"; + public static final String SQ_QUEUE_NAME = "SequentialByOriginator"; + public static final String SQ_QUEUE_TOPIC = "tb_rule_engine.sq"; + } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/QueueKey.java b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/QueueKey.java index e43da75869..43d741fc02 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/QueueKey.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/QueueKey.java @@ -17,6 +17,7 @@ package org.thingsboard.server.queue.discovery; import lombok.AllArgsConstructor; import lombok.Data; +import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.queue.Queue; import org.thingsboard.server.common.msg.queue.ServiceType; @@ -24,7 +25,6 @@ import org.thingsboard.server.common.msg.queue.ServiceType; @Data @AllArgsConstructor public class QueueKey { - private static final String MAIN = "Main"; private final ServiceType type; private final String queueName; @@ -44,13 +44,13 @@ public class QueueKey { public QueueKey(ServiceType type, TenantId tenantId) { this.type = type; - this.queueName = MAIN; + this.queueName = DataConstants.MAIN_QUEUE_NAME; this.tenantId = tenantId != null ? tenantId : TenantId.SYS_TENANT_ID; } public QueueKey(ServiceType type) { this.type = type; - this.queueName = MAIN; + this.queueName = DataConstants.MAIN_QUEUE_NAME; this.tenantId = TenantId.SYS_TENANT_ID; } diff --git a/common/queue/src/test/java/org/thingsboard/server/queue/discovery/QueueKeyTest.java b/common/queue/src/test/java/org/thingsboard/server/queue/discovery/QueueKeyTest.java index 6fcb365f09..be2e2e2750 100644 --- a/common/queue/src/test/java/org/thingsboard/server/queue/discovery/QueueKeyTest.java +++ b/common/queue/src/test/java/org/thingsboard/server/queue/discovery/QueueKeyTest.java @@ -17,6 +17,7 @@ package org.thingsboard.server.queue.discovery; import lombok.extern.slf4j.Slf4j; import org.junit.jupiter.api.Test; +import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.msg.queue.ServiceType; @@ -29,7 +30,7 @@ class QueueKeyTest { @Test void testToStringSystemTenant() { - QueueKey queueKey = new QueueKey(ServiceType.TB_RULE_ENGINE, "Main", TenantId.SYS_TENANT_ID); + QueueKey queueKey = new QueueKey(ServiceType.TB_RULE_ENGINE, DataConstants.MAIN_QUEUE_NAME, TenantId.SYS_TENANT_ID); log.info("The queue key is {}",queueKey); assertThat(queueKey.toString()).isEqualTo("QK(Main,TB_RULE_ENGINE,system)"); } @@ -37,7 +38,7 @@ class QueueKeyTest { @Test void testToStringCustomTenant() { TenantId tenantId = TenantId.fromUUID(UUID.fromString("3ebd39eb-43d4-4911-a818-cdbf8d508f88")); - QueueKey queueKey = new QueueKey(ServiceType.TB_RULE_ENGINE, "Main", tenantId); + QueueKey queueKey = new QueueKey(ServiceType.TB_RULE_ENGINE, DataConstants.MAIN_QUEUE_NAME, tenantId); log.info("The queue key is {}",queueKey); assertThat(queueKey.toString()).isEqualTo("QK(Main,TB_RULE_ENGINE,3ebd39eb-43d4-4911-a818-cdbf8d508f88)"); } diff --git a/dao/src/main/java/org/thingsboard/server/dao/service/validator/TenantProfileDataValidator.java b/dao/src/main/java/org/thingsboard/server/dao/service/validator/TenantProfileDataValidator.java index 0368e38173..577597be2a 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/service/validator/TenantProfileDataValidator.java +++ b/dao/src/main/java/org/thingsboard/server/dao/service/validator/TenantProfileDataValidator.java @@ -18,6 +18,7 @@ package org.thingsboard.server.dao.service.validator; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Lazy; import org.springframework.stereotype.Component; +import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.StringUtils; import org.thingsboard.server.common.data.TenantProfile; import org.thingsboard.server.common.data.id.TenantId; @@ -72,7 +73,7 @@ public class TenantProfileDataValidator extends DataValidator { Optional mainQueueConfig = queueConfiguration .stream() - .filter(q -> q.getName().equals("Main")) + .filter(q -> q.getName().equals(DataConstants.MAIN_QUEUE_NAME)) .findAny(); if (mainQueueConfig.isEmpty()) { throw new DataValidationException("Main queue configuration should be specified!"); diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/queue/JpaQueueDao.java b/dao/src/main/java/org/thingsboard/server/dao/sql/queue/JpaQueueDao.java index f11952f3e9..3ffd137871 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/queue/JpaQueueDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/queue/JpaQueueDao.java @@ -20,6 +20,7 @@ import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.data.jpa.repository.JpaRepository; import org.springframework.stereotype.Component; +import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.page.PageData; import org.thingsboard.server.common.data.page.PageLink; @@ -70,7 +71,7 @@ public class JpaQueueDao extends JpaAbstractDao implements Q @Override public List findAllMainQueues() { - List entities = Lists.newArrayList(queueRepository.findAllByName("Main")); + List entities = Lists.newArrayList(queueRepository.findAllByName(DataConstants.MAIN_QUEUE_NAME)); return DaoUtil.convertDataList(entities); } diff --git a/dao/src/test/java/org/thingsboard/server/dao/service/BaseQueueServiceTest.java b/dao/src/test/java/org/thingsboard/server/dao/service/BaseQueueServiceTest.java index c2365ab1e9..783f67124f 100644 --- a/dao/src/test/java/org/thingsboard/server/dao/service/BaseQueueServiceTest.java +++ b/dao/src/test/java/org/thingsboard/server/dao/service/BaseQueueServiceTest.java @@ -19,6 +19,7 @@ import org.junit.After; import org.junit.Assert; import org.junit.Before; import org.junit.Test; +import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.Tenant; import org.thingsboard.server.common.data.TenantProfile; import org.thingsboard.server.common.data.id.TenantId; @@ -54,8 +55,8 @@ public abstract class BaseQueueServiceTest extends AbstractServiceTest { tenantProfile.setIsolatedTbRuleEngine(true); TenantProfileQueueConfiguration mainQueueConfiguration = new TenantProfileQueueConfiguration(); - mainQueueConfiguration.setName("Main"); - mainQueueConfiguration.setTopic("tb_rule_engine.main"); + mainQueueConfiguration.setName(DataConstants.MAIN_QUEUE_NAME); + mainQueueConfiguration.setTopic(DataConstants.MAIN_QUEUE_TOPIC); mainQueueConfiguration.setPollInterval(25); mainQueueConfiguration.setPartitions(10); mainQueueConfiguration.setConsumerPerPartition(true); @@ -495,7 +496,7 @@ public abstract class BaseQueueServiceTest extends AbstractServiceTest { for (int i = 0; i < loadedQueues.size(); i++) { Queue queue = loadedQueues.get(i); - if (queue.getName().equals("Main")) { + if (queue.getName().equals(DataConstants.MAIN_QUEUE_NAME)) { loadedQueues.remove(queue); break; } diff --git a/dao/src/test/java/org/thingsboard/server/dao/service/BaseTenantProfileServiceTest.java b/dao/src/test/java/org/thingsboard/server/dao/service/BaseTenantProfileServiceTest.java index f245af61c1..f96bb85016 100644 --- a/dao/src/test/java/org/thingsboard/server/dao/service/BaseTenantProfileServiceTest.java +++ b/dao/src/test/java/org/thingsboard/server/dao/service/BaseTenantProfileServiceTest.java @@ -19,6 +19,7 @@ import com.fasterxml.jackson.databind.node.NullNode; import org.junit.After; import org.junit.Assert; import org.junit.Test; +import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.EntityInfo; import org.thingsboard.server.common.data.Tenant; import org.thingsboard.server.common.data.TenantProfile; @@ -57,8 +58,8 @@ public abstract class BaseTenantProfileServiceTest extends AbstractServiceTest { tenantProfile.setIsolatedTbRuleEngine(true); TenantProfileQueueConfiguration mainQueueConfiguration = new TenantProfileQueueConfiguration(); - mainQueueConfiguration.setName("Main"); - mainQueueConfiguration.setTopic("tb_rule_engine.main"); + mainQueueConfiguration.setName(DataConstants.MAIN_QUEUE_NAME); + mainQueueConfiguration.setTopic(DataConstants.MAIN_QUEUE_TOPIC); mainQueueConfiguration.setPollInterval(25); mainQueueConfiguration.setPartitions(10); mainQueueConfiguration.setConsumerPerPartition(true); @@ -295,8 +296,8 @@ public abstract class BaseTenantProfileServiceTest extends AbstractServiceTest { private void addMainQueueConfig(TenantProfile tenantProfile) { TenantProfileQueueConfiguration mainQueueConfiguration = new TenantProfileQueueConfiguration(); - mainQueueConfiguration.setName("Main"); - mainQueueConfiguration.setTopic("tb_rule_engine.main"); + mainQueueConfiguration.setName(DataConstants.MAIN_QUEUE_NAME); + mainQueueConfiguration.setTopic(DataConstants.MAIN_QUEUE_TOPIC); mainQueueConfiguration.setPollInterval(25); mainQueueConfiguration.setPartitions(10); mainQueueConfiguration.setConsumerPerPartition(true); diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleNodeCacheService.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleNodeCacheService.java new file mode 100644 index 0000000000..d627923168 --- /dev/null +++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleNodeCacheService.java @@ -0,0 +1,46 @@ +/** + * Copyright © 2016-2022 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.rule.engine.api; + +import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.msg.TbMsg; + +import java.util.List; +import java.util.Set; + +public interface RuleNodeCacheService { + + void add(String key, String value); + + void add(String key, EntityId id); + + void add(String key, TbMsg value); + + void removeStringList(String key, List values); + + void removeEntityIdList(String key, List values); + + void removeTbMsgList(String key, List values); + + Set getStrings(String key); + + Set getEntityIds(String key); + + Set getTbMsgs(String key, String queueName); + + void evict(String key); + +} diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java index 1b34fbc95d..e7c0943188 100644 --- a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java +++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java @@ -317,6 +317,8 @@ public interface TbContext { void clearRuleNodeStates(); + RuleNodeCacheService getRuleNodeCacheService(); + void addTenantProfileListener(Consumer listener); void addDeviceProfileListeners(Consumer listener, BiConsumer deviceListener); diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbNode.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbNode.java index 1516b6957c..e8641dafe1 100644 --- a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbNode.java +++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbNode.java @@ -31,6 +31,11 @@ public interface TbNode { default void destroy() {} + // todo: add reason for destroy + default void destroy(TbContext ctx) { + destroy(); + } + default void onPartitionChangeMsg(TbContext ctx, PartitionChangeMsg msg) {} } diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbNodeException.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbNodeException.java index 4e1e875d28..470e37e509 100644 --- a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbNodeException.java +++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbNodeException.java @@ -15,8 +15,6 @@ */ package org.thingsboard.rule.engine.api; -import com.fasterxml.jackson.core.JsonProcessingException; - /** * Created by ashvayka on 19.01.18. */ diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/deduplication/DeduplicationData.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/deduplication/DeduplicationData.java index 58f170e0ab..501079658c 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/deduplication/DeduplicationData.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/deduplication/DeduplicationData.java @@ -39,6 +39,10 @@ public class DeduplicationData { msgList.add(msg); } + public void addAll(List tbMsgs) { + msgList.addAll(tbMsgs); + } + public boolean isEmpty() { return msgList.isEmpty(); } diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/deduplication/TbMsgDeduplicationNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/deduplication/TbMsgDeduplicationNode.java index b63894a61e..2bea4cfa69 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/deduplication/TbMsgDeduplicationNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/deduplication/TbMsgDeduplicationNode.java @@ -39,6 +39,7 @@ import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.Optional; +import java.util.Set; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; @@ -59,6 +60,7 @@ import java.util.concurrent.TimeUnit; public class TbMsgDeduplicationNode implements TbNode { private static final String TB_MSG_DEDUPLICATION_TIMEOUT_MSG = "TbMsgDeduplicationNodeMsg"; + private static final String DEDUPLICATION_IDS_CACHE_KEY = "deduplication_ids"; public static final int TB_MSG_DEDUPLICATION_RETRY_DELAY = 10; private static final String EMPTY_DATA = ""; private static final TbMsgMetaData EMPTY_META_DATA = new TbMsgMetaData(); @@ -78,6 +80,9 @@ public class TbMsgDeduplicationNode implements TbNode { this.config = TbNodeUtils.convert(configuration, TbMsgDeduplicationNodeConfiguration.class); this.deduplicationInterval = TimeUnit.SECONDS.toMillis(config.getInterval()); this.deduplicationId = config.getId(); + if (deduplicationMap.isEmpty()) { + getDeduplicationDataFromCacheAndSchedule(ctx); + } } @Override @@ -90,8 +95,36 @@ public class TbMsgDeduplicationNode implements TbNode { } @Override - public void destroy() { + public void destroy(TbContext ctx) { deduplicationMap.clear(); + Set deduplicationIds = getDeduplicationIds(ctx); + if (deduplicationIds == null) { + return; + } + deduplicationIds.forEach(id -> ctx.getRuleNodeCacheService().evict(id.toString())); + ctx.getRuleNodeCacheService().evict(DEDUPLICATION_IDS_CACHE_KEY); + } + + private void getDeduplicationDataFromCacheAndSchedule(TbContext ctx) { + Set deduplicationIds = getDeduplicationIds(ctx); + if (deduplicationIds == null) { + return; + } + deduplicationIds.forEach(id -> { + List tbMsgs = new ArrayList<>(ctx.getRuleNodeCacheService().getTbMsgs(id.toString(), config.getQueueName())); + DeduplicationData deduplicationData = new DeduplicationData(); + deduplicationData.addAll(tbMsgs); + deduplicationMap.put(id, deduplicationData); + scheduleTickMsg(ctx, id, deduplicationData); + }); + } + + private Set getDeduplicationIds(TbContext ctx) { + Set deduplicationIds = ctx.getRuleNodeCacheService().getEntityIds(TbMsgDeduplicationNode.DEDUPLICATION_IDS_CACHE_KEY); + if (deduplicationIds.isEmpty()) { + return null; + } + return deduplicationIds; } private void processOnRegularMsg(TbContext ctx, TbMsg msg) { @@ -99,7 +132,11 @@ public class TbMsgDeduplicationNode implements TbNode { DeduplicationData deduplicationMsgs = deduplicationMap.computeIfAbsent(id, k -> new DeduplicationData()); if (deduplicationMsgs.size() < config.getMaxPendingMsgs()) { log.trace("[{}][{}] Adding msg: [{}][{}] to the pending msgs map ...", ctx.getSelfId(), id, msg.getId(), msg.getMetaDataTs()); + if (deduplicationMsgs.isEmpty()) { + ctx.getRuleNodeCacheService().add(DEDUPLICATION_IDS_CACHE_KEY, id); + } deduplicationMsgs.add(msg); + ctx.getRuleNodeCacheService().add(id.toString(), msg); ctx.ack(msg); scheduleTickMsg(ctx, id, deduplicationMsgs); } else { @@ -132,13 +169,13 @@ public class TbMsgDeduplicationNode implements TbNode { } long deduplicationTimeoutMs = System.currentTimeMillis(); try { - List deduplicationResults = new ArrayList<>(); + List>> deduplicationResults = new ArrayList<>(); List msgList = data.getMsgList(); Optional> packBoundsOpt = findValidPack(msgList, deduplicationTimeoutMs); while (packBoundsOpt.isPresent()) { TbPair packBounds = packBoundsOpt.get(); + List pack = new ArrayList<>(); if (DeduplicationStrategy.ALL.equals(config.getStrategy())) { - List pack = new ArrayList<>(); for (Iterator iterator = msgList.iterator(); iterator.hasNext(); ) { TbMsg msg = iterator.next(); long msgTs = msg.getMetaDataTs(); @@ -147,12 +184,12 @@ public class TbMsgDeduplicationNode implements TbNode { iterator.remove(); } } - deduplicationResults.add(TbMsg.newMsg( + deduplicationResults.add(new TbPair<>(TbMsg.newMsg( config.getQueueName(), config.getOutMsgType(), deduplicationId, - getMetadata(), - getMergedData(pack))); + getMetadata(packBounds.getFirst()), + getMergedData(pack)), pack)); } else { TbMsg resultMsg = null; boolean searchMin = DeduplicationStrategy.FIRST.equals(config.getStrategy()); @@ -160,6 +197,7 @@ public class TbMsgDeduplicationNode implements TbNode { TbMsg msg = iterator.next(); long msgTs = msg.getMetaDataTs(); if (msgTs >= packBounds.getFirst() && msgTs < packBounds.getSecond()) { + pack.add(msg); iterator.remove(); if (resultMsg == null || (searchMin && msg.getMetaDataTs() < resultMsg.getMetaDataTs()) @@ -168,11 +206,14 @@ public class TbMsgDeduplicationNode implements TbNode { } } } - deduplicationResults.add(resultMsg); + if (resultMsg != null) { + resultMsg = TbMsg.transformMsg(resultMsg, config.getQueueName()); + deduplicationResults.add(new TbPair<>(resultMsg, pack)); + } } packBoundsOpt = findValidPack(msgList, deduplicationTimeoutMs); } - deduplicationResults.forEach(outMsg -> enqueueForTellNextWithRetry(ctx, outMsg, 0)); + deduplicationResults.forEach(result -> enqueueForTellNextWithRetry(ctx, result, 0)); } finally { if (!data.isEmpty()) { scheduleTickMsg(ctx, deduplicationId, data); @@ -199,17 +240,18 @@ public class TbMsgDeduplicationNode implements TbNode { }); } - private void enqueueForTellNextWithRetry(TbContext ctx, TbMsg msg, int retryAttempt) { + private void enqueueForTellNextWithRetry(TbContext ctx, TbPair> result, int retryAttempt) { if (config.getMaxRetries() > retryAttempt) { - ctx.enqueueForTellNext(msg, TbRelationTypes.SUCCESS, + TbMsg outMsg = result.getFirst(); + List msgsToRemoveFromCache = result.getSecond(); + ctx.enqueueForTellNext(outMsg, TbRelationTypes.SUCCESS, () -> { - log.trace("[{}][{}][{}] Successfully enqueue deduplication result message!", ctx.getSelfId(), msg.getOriginator(), retryAttempt); + log.trace("[{}][{}][{}] Successfully enqueue deduplication result message!", ctx.getSelfId(), outMsg.getOriginator(), retryAttempt); + ctx.getRuleNodeCacheService().removeTbMsgList(outMsg.getOriginator().toString(), msgsToRemoveFromCache); }, throwable -> { - log.trace("[{}][{}][{}] Failed to enqueue deduplication output message due to: ", ctx.getSelfId(), msg.getOriginator(), retryAttempt, throwable); - ctx.schedule(() -> { - enqueueForTellNextWithRetry(ctx, msg, retryAttempt + 1); - }, TB_MSG_DEDUPLICATION_RETRY_DELAY, TimeUnit.SECONDS); + log.trace("[{}][{}][{}] Failed to enqueue deduplication output message due to: ", ctx.getSelfId(), outMsg.getOriginator(), retryAttempt, throwable); + ctx.schedule(() -> enqueueForTellNextWithRetry(ctx, result, retryAttempt + 1), TB_MSG_DEDUPLICATION_RETRY_DELAY, TimeUnit.SECONDS); }); } } @@ -229,9 +271,9 @@ public class TbMsgDeduplicationNode implements TbNode { return JacksonUtil.toString(mergedData); } - private TbMsgMetaData getMetadata() { + private TbMsgMetaData getMetadata(long packStartTs) { TbMsgMetaData metaData = new TbMsgMetaData(); - metaData.putValue("ts", String.valueOf(System.currentTimeMillis())); + metaData.putValue("ts", String.valueOf(packStartTs)); return metaData; } diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/deduplication/TbMsgDeduplicationNodeConfiguration.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/deduplication/TbMsgDeduplicationNodeConfiguration.java index afb9cc62b6..93612e5ac2 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/deduplication/TbMsgDeduplicationNodeConfiguration.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/deduplication/TbMsgDeduplicationNodeConfiguration.java @@ -17,6 +17,7 @@ package org.thingsboard.rule.engine.deduplication; import lombok.Data; import org.thingsboard.rule.engine.api.NodeConfiguration; +import org.thingsboard.server.common.data.DataConstants; @Data public class TbMsgDeduplicationNodeConfiguration implements NodeConfiguration { @@ -28,10 +29,10 @@ public class TbMsgDeduplicationNodeConfiguration implements NodeConfiguration) invocationOnMock -> { String type = (String) (invocationOnMock.getArguments())[1]; @@ -171,11 +185,23 @@ public class TbMsgDeduplicationNodeTest { ArgumentCaptor successCaptor = ArgumentCaptor.forClass(Runnable.class); ArgumentCaptor> failureCaptor = ArgumentCaptor.forClass(Consumer.class); + verify(ruleNodeCacheService, times(1)).getEntityIds(anyString()); + verify(ruleNodeCacheService, times(1)).add(eq(DEDUPLICATION_IDS_CACHE_KEY), eq(deviceId)); + verify(ruleNodeCacheService, times(msgCount)).add(eq(deviceId.toString()), any(TbMsg.class)); + + verify(node, times(msgCount + wantedNumberOfTellSelfInvocation + 1)).onMsg(eq(ctx), any()); + verify(ctx, times(msgCount)).ack(any()); verify(ctx, times(1)).tellFailure(eq(msgToReject), any()); - verify(node, times(msgCount + wantedNumberOfTellSelfInvocation + 1)).onMsg(eq(ctx), any()); verify(ctx, times(1)).enqueueForTellNext(newMsgCaptor.capture(), eq(TbRelationTypes.SUCCESS), successCaptor.capture(), failureCaptor.capture()); - Assertions.assertEquals(inputMsgs.get(0), newMsgCaptor.getValue()); + + for (Runnable valueCaptor : successCaptor.getAllValues()) { + valueCaptor.run(); + } + verify(ruleNodeCacheService, times(1)).removeTbMsgList(eq(deviceId.toString()), anyList()); + verify(ctx, never()).schedule(any(), anyLong(), any()); + + Assertions.assertEquals(TbMsg.toByteString(inputMsgs.get(0)), TbMsg.toByteString(newMsgCaptor.getValue())); } @Test @@ -210,11 +236,23 @@ public class TbMsgDeduplicationNodeTest { ArgumentCaptor successCaptor = ArgumentCaptor.forClass(Runnable.class); ArgumentCaptor> failureCaptor = ArgumentCaptor.forClass(Consumer.class); + verify(ruleNodeCacheService, times(1)).getEntityIds(anyString()); + verify(ruleNodeCacheService, times(1)).add(eq(DEDUPLICATION_IDS_CACHE_KEY), eq(deviceId)); + verify(ruleNodeCacheService, times(msgCount)).add(eq(deviceId.toString()), any(TbMsg.class)); + + verify(node, times(msgCount + wantedNumberOfTellSelfInvocation + 1)).onMsg(eq(ctx), any()); + verify(ctx, times(msgCount)).ack(any()); verify(ctx, times(1)).tellFailure(eq(msgToReject), any()); - verify(node, times(msgCount + wantedNumberOfTellSelfInvocation + 1)).onMsg(eq(ctx), any()); verify(ctx, times(1)).enqueueForTellNext(newMsgCaptor.capture(), eq(TbRelationTypes.SUCCESS), successCaptor.capture(), failureCaptor.capture()); - Assertions.assertEquals(msgWithLatestTs, newMsgCaptor.getValue()); + + for (Runnable valueCaptor : successCaptor.getAllValues()) { + valueCaptor.run(); + } + verify(ruleNodeCacheService, times(1)).removeTbMsgList(eq(deviceId.toString()), anyList()); + verify(ctx, never()).schedule(any(), anyLong(), any()); + + Assertions.assertEquals(TbMsg.toByteString(msgWithLatestTs), TbMsg.toByteString(newMsgCaptor.getValue())); } @Test @@ -227,7 +265,7 @@ public class TbMsgDeduplicationNodeTest { config.setInterval(deduplicationInterval); config.setStrategy(DeduplicationStrategy.ALL); config.setOutMsgType(SessionMsgType.POST_ATTRIBUTES_REQUEST.name()); - config.setQueueName(HIGH_PRIORITY_QUEUE_NAME); + config.setQueueName(DataConstants.HP_QUEUE_NAME); nodeConfiguration = new TbNodeConfiguration(JacksonUtil.valueToTree(config)); node.init(ctx, nodeConfiguration); @@ -245,10 +283,21 @@ public class TbMsgDeduplicationNodeTest { ArgumentCaptor successCaptor = ArgumentCaptor.forClass(Runnable.class); ArgumentCaptor> failureCaptor = ArgumentCaptor.forClass(Consumer.class); - verify(ctx, times(msgCount)).ack(any()); + verify(ruleNodeCacheService, times(1)).getEntityIds(anyString()); + verify(ruleNodeCacheService, times(1)).add(eq(DEDUPLICATION_IDS_CACHE_KEY), eq(deviceId)); + verify(ruleNodeCacheService, times(msgCount)).add(eq(deviceId.toString()), any(TbMsg.class)); + verify(node, times(msgCount + wantedNumberOfTellSelfInvocation)).onMsg(eq(ctx), any()); + + verify(ctx, times(msgCount)).ack(any()); verify(ctx, times(1)).enqueueForTellNext(newMsgCaptor.capture(), eq(TbRelationTypes.SUCCESS), successCaptor.capture(), failureCaptor.capture()); + for (Runnable valueCaptor : successCaptor.getAllValues()) { + valueCaptor.run(); + } + verify(ruleNodeCacheService, times(1)).removeTbMsgList(eq(deviceId.toString()), anyList()); + verify(ctx, never()).schedule(any(), anyLong(), any()); + Assertions.assertEquals(1, newMsgCaptor.getAllValues().size()); TbMsg outMessage = newMsgCaptor.getAllValues().get(0); Assertions.assertEquals(getMergedData(inputMsgs), outMessage.getData()); @@ -267,7 +316,7 @@ public class TbMsgDeduplicationNodeTest { config.setInterval(deduplicationInterval); config.setStrategy(DeduplicationStrategy.ALL); config.setOutMsgType(SessionMsgType.POST_ATTRIBUTES_REQUEST.name()); - config.setQueueName(HIGH_PRIORITY_QUEUE_NAME); + config.setQueueName(DataConstants.HP_QUEUE_NAME); nodeConfiguration = new TbNodeConfiguration(JacksonUtil.valueToTree(config)); node.init(ctx, nodeConfiguration); @@ -278,7 +327,7 @@ public class TbMsgDeduplicationNodeTest { for (TbMsg msg : firstMsgPack) { node.onMsg(ctx, msg); } - long firstPackDeduplicationPackEndTs = firstMsgPack.get(0).getMetaDataTs() + TimeUnit.SECONDS.toMillis(deduplicationInterval); + long firstPackDeduplicationPackEndTs = firstMsgPack.get(0).getMetaDataTs() + TimeUnit.SECONDS.toMillis(deduplicationInterval); List secondMsgPack = getTbMsgs(deviceId, msgCount / 2, firstPackDeduplicationPackEndTs, 500); for (TbMsg msg : secondMsgPack) { @@ -291,10 +340,21 @@ public class TbMsgDeduplicationNodeTest { ArgumentCaptor successCaptor = ArgumentCaptor.forClass(Runnable.class); ArgumentCaptor> failureCaptor = ArgumentCaptor.forClass(Consumer.class); - verify(ctx, times(msgCount)).ack(any()); + verify(ruleNodeCacheService, times(1)).getEntityIds(anyString()); + verify(ruleNodeCacheService, times(1)).add(eq(DEDUPLICATION_IDS_CACHE_KEY), eq(deviceId)); + verify(ruleNodeCacheService, times(msgCount)).add(eq(deviceId.toString()), any(TbMsg.class)); + verify(node, times(msgCount + wantedNumberOfTellSelfInvocation)).onMsg(eq(ctx), any()); + + verify(ctx, times(msgCount)).ack(any()); verify(ctx, times(2)).enqueueForTellNext(newMsgCaptor.capture(), eq(TbRelationTypes.SUCCESS), successCaptor.capture(), failureCaptor.capture()); + for (Runnable valueCaptor : successCaptor.getAllValues()) { + valueCaptor.run(); + } + verify(ruleNodeCacheService, times(2)).removeTbMsgList(eq(deviceId.toString()), anyList()); + verify(ctx, never()).schedule(any(), anyLong(), any()); + List resultMsgs = newMsgCaptor.getAllValues(); Assertions.assertEquals(2, resultMsgs.size()); @@ -345,14 +405,26 @@ public class TbMsgDeduplicationNodeTest { ArgumentCaptor successCaptor = ArgumentCaptor.forClass(Runnable.class); ArgumentCaptor> failureCaptor = ArgumentCaptor.forClass(Consumer.class); - verify(ctx, times(msgCount)).ack(any()); + verify(ruleNodeCacheService, times(1)).getEntityIds(anyString()); + verify(ruleNodeCacheService, times(1)).add(eq(DEDUPLICATION_IDS_CACHE_KEY), eq(deviceId)); + verify(ruleNodeCacheService, times(msgCount)).add(eq(deviceId.toString()), any(TbMsg.class)); + verify(node, times(msgCount + wantedNumberOfTellSelfInvocation)).onMsg(eq(ctx), any()); + + verify(ctx, times(msgCount)).ack(any()); verify(ctx, times(2)).enqueueForTellNext(newMsgCaptor.capture(), eq(TbRelationTypes.SUCCESS), successCaptor.capture(), failureCaptor.capture()); + for (Runnable valueCaptor : successCaptor.getAllValues()) { + valueCaptor.run(); + } + verify(ruleNodeCacheService, times(2)).removeTbMsgList(eq(deviceId.toString()), anyList()); + verify(ctx, never()).schedule(any(), anyLong(), any()); + List resultMsgs = newMsgCaptor.getAllValues(); Assertions.assertEquals(2, resultMsgs.size()); - Assertions.assertTrue(resultMsgs.contains(msgWithLatestTsInFirstPack)); - Assertions.assertTrue(resultMsgs.contains(msgWithLatestTsInSecondPack)); + List resultMsgsByteStrings = resultMsgs.stream().map(TbMsg::toByteString).collect(Collectors.toList()); + Assertions.assertTrue(resultMsgsByteStrings.contains(TbMsg.toByteString(msgWithLatestTsInFirstPack))); + Assertions.assertTrue(resultMsgsByteStrings.contains(TbMsg.toByteString(msgWithLatestTsInSecondPack))); } private TbMsg getMsgWithLatestTs(List firstMsgPack) { @@ -381,7 +453,7 @@ public class TbMsgDeduplicationNodeTest { TbMsgMetaData metaData = new TbMsgMetaData(); metaData.putValue("ts", String.valueOf(ts)); return TbMsg.newMsg( - MAIN_QUEUE_NAME, + DataConstants.MAIN_QUEUE_NAME, SessionMsgType.POST_TELEMETRY_REQUEST.name(), deviceId, metaData,