From 2f26b8dbc0c01465152a2e343150bff07a2f3148 Mon Sep 17 00:00:00 2001 From: ShvaykaD Date: Tue, 24 Jan 2023 17:56:32 +0200 Subject: [PATCH] refactoring after review --- .../server/actors/ActorSystemContext.java | 4 +- .../actors/ruleChain/DefaultTbContext.java | 58 +-------- .../DefaultTbRuleNodeCacheService.java | 90 ++++++++++++++ .../RuleNodeActorMessageProcessor.java | 3 +- .../rule/DefaultRuleNodeCacheService.java | 95 --------------- .../service/rule/TbRuleNodeCacheService.java | 22 ++++ common/cache/pom.xml | 4 + .../cache/rule/DefaultRuleNodeCache.java | 49 +++++--- .../server/cache/rule/RedisRuleNodeCache.java | 114 ++++++++++++++++-- .../server/cache/rule/RuleNodeCache.java | 23 +++- .../engine/api}/RuleNodeCacheService.java | 25 ++-- .../rule/engine/api/TbContext.java | 22 +--- .../thingsboard/rule/engine/api/TbNode.java | 7 +- .../rule/engine/api/TbNodeException.java | 2 - .../deduplication/TbMsgDeduplicationNode.java | 21 ++-- .../action/TbMsgDeduplicationNodeTest.java | 80 +++++++++++- 16 files changed, 380 insertions(+), 239 deletions(-) create mode 100644 application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbRuleNodeCacheService.java delete mode 100644 application/src/main/java/org/thingsboard/server/service/rule/DefaultRuleNodeCacheService.java create mode 100644 application/src/main/java/org/thingsboard/server/service/rule/TbRuleNodeCacheService.java rename {application/src/main/java/org/thingsboard/server/service/rule => rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api}/RuleNodeCacheService.java (52%) 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 511211c303..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; @@ -96,7 +97,6 @@ import org.thingsboard.server.service.profile.TbDeviceProfileCache; import org.thingsboard.server.service.rpc.TbCoreDeviceRpcService; import org.thingsboard.server.service.rpc.TbRpcService; import org.thingsboard.server.service.rpc.TbRuleEngineDeviceRpcService; -import org.thingsboard.server.service.rule.RuleNodeCacheService; import org.thingsboard.server.service.session.DeviceSessionCacheService; import org.thingsboard.server.service.sms.SmsExecutorService; import org.thingsboard.server.service.state.DeviceStateService; @@ -237,7 +237,7 @@ public class ActorSystemContext { @Autowired @Getter - private RuleNodeCacheService ruleNodeCacheService; + 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 398d9c3a8a..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 @@ -734,58 +738,8 @@ class DefaultTbContext implements TbContext { } @Override - public void addToRuleNodeCache(String key, TbMsg msg) { - mainCtx.getRuleNodeCacheService().add(this.getSelfId(), key, msg); - } - - @Override - public void addToRuleNodeCache(String key, List tbMsgList) { - mainCtx.getRuleNodeCacheService().add(this.getSelfId(), key, tbMsgList); - } - - @Override - public void addToRuleNodeCache(String key, byte[]... values) { - mainCtx.getRuleNodeCacheService().add(this.getSelfId(), key, values); - } - - @Override - public void addToRuleNodeCache(String key, String value) { - mainCtx.getRuleNodeCacheService().add(this.getSelfId(), key, value.getBytes()); - } - - @Override - public Set getFromRuleNodeCache(String key) { - return mainCtx.getRuleNodeCacheService().get(this.getSelfId(), key); - } - - @Override - public Set getFromRuleNodeCache(String key, String queueName) { - return mainCtx.getRuleNodeCacheService().get(this.getSelfId(), key, queueName); - } - - @Override - public void removeFromRuleNodeCache(String key, TbMsg msg) { - mainCtx.getRuleNodeCacheService().remove(this.getSelfId(), key, msg); - } - - @Override - public void removeFromRuleNodeCache(String key, List tbMsgList) { - mainCtx.getRuleNodeCacheService().remove(this.getSelfId(), key, tbMsgList); - } - - @Override - public void removeFromRuleNodeCache(String key, byte[]... values) { - mainCtx.getRuleNodeCacheService().remove(this.getSelfId(), key, values); - } - - @Override - public void removeFromRuleNodeCache(String key, String value) { - mainCtx.getRuleNodeCacheService().remove(this.getSelfId(), key, value.getBytes()); - } - - @Override - public void evictFromRuleNodeCache(String key) { - mainCtx.getRuleNodeCacheService().evict(this.getSelfId(), key); + public RuleNodeCacheService getRuleNodeCacheService() { + return ruleNodeCacheService; } @Override 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 06c5e9e7e1..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,8 +88,7 @@ public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor values) { - ruleNodeCache.add(toRuleNodeCacheKey(ruleNodeId, key), tbMsgListToBytes(values)); - } - - @Override - public Set get(RuleNodeId ruleNodeId, String key) { - return ruleNodeCache.get(toRuleNodeCacheKey(ruleNodeId, key)); - } - - @Override - public Set get(RuleNodeId ruleNodeId, String key, String queueName) { - return ruleNodeCache.get(toRuleNodeCacheKey(ruleNodeId, key)) - .stream() - .map(bytes -> TbMsg.fromBytes(queueName, bytes, TbMsgCallback.EMPTY)) - .collect(Collectors.toSet()); - } - - @Override - public void remove(RuleNodeId ruleNodeId, String key, byte[]... values) { - ruleNodeCache.remove(toRuleNodeCacheKey(ruleNodeId, key), values); - } - - @Override - public void remove(RuleNodeId ruleNodeId, String key, TbMsg value) { - ruleNodeCache.remove(toRuleNodeCacheKey(ruleNodeId, key), TbMsg.toByteArray(value)); - } - - @Override - public void remove(RuleNodeId ruleNodeId, String key, List values) { - ruleNodeCache.remove(toRuleNodeCacheKey(ruleNodeId, key), tbMsgListToBytes(values)); - } - - @Override - public void evict(RuleNodeId ruleNodeId, String key) { - ruleNodeCache.evict(toRuleNodeCacheKey(ruleNodeId, key)); - } - - private String toRuleNodeCacheKey(RuleNodeId ruleNodeId, String key) { - return String.format("%s::%s", ruleNodeId.toString(), key); - } - - private byte[][] tbMsgListToBytes(List values) { - return values.stream() - .map(TbMsg::toByteArray) - .toArray(byte[][]::new); - } - -} 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/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 index 9631df5e08..8cbdc4e415 100644 --- 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 @@ -15,49 +15,60 @@ */ package org.thingsboard.server.cache.rule; -import lombok.extern.slf4j.Slf4j; 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.Arrays; import java.util.Collections; -import java.util.HashMap; -import java.util.Map; +import java.util.List; import java.util.Set; -import java.util.TreeSet; @Service @ConditionalOnProperty(prefix = "cache", value = "type", havingValue = "caffeine", matchIfMissing = true) public class DefaultRuleNodeCache implements RuleNodeCache { - private final Map> cache; + @Override + public void add(String key, String value) { + } + + @Override + public void add(String key, EntityId value) { + } + + @Override + public void add(String key, TbMsg value) { + } - public DefaultRuleNodeCache() { - cache = new HashMap<>(); + @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 void add(String key, byte[]... values) { - Set data = cache.computeIfAbsent(key, s -> new TreeSet<>(Arrays::compare)); - data.addAll(Arrays.asList(values)); + public Set getStringSetByKey(String key) { + return Collections.emptySet(); } @Override - public void remove(String key, byte[]... values) { - cache.computeIfPresent(key, (id, data) -> { - Arrays.asList(values).forEach(data::remove); - return data; - }); + public Set getEntityIdSetByKey(String key) { + return Collections.emptySet(); } @Override - public Set get(String key) { - return cache.getOrDefault(key, Collections.emptySet()); + public Set getTbMsgSetByKey(String key, String queueName) { + return Collections.emptySet(); } @Override public void evict(String key) { - cache.remove(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 index bc583074b0..49b7ff67bb 100644 --- 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 @@ -20,8 +20,15 @@ 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") @@ -31,31 +38,116 @@ public class RedisRuleNodeCache implements RuleNodeCache { private final RedisConnectionFactory redisConnectionFactory; @Override - public void add(String key, byte[]... values) { + 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.setCommands().sAdd(key.getBytes(), values); + connection.del(key.getBytes()); } } - @Override - public void remove(String key, byte[]... values) { + private void processAdd(String key, byte[] value) { try (RedisConnection connection = redisConnectionFactory.getConnection()) { - connection.setCommands().sRem(key.getBytes(), values); + connection.setCommands().sAdd(key.getBytes(), value); } } - @Override - public Set get(String key) { + private void processRemove(String key, byte[][] values) { + if (values.length == 0) { + return; + } try (RedisConnection connection = redisConnectionFactory.getConnection()) { - return connection.setCommands().sMembers(key.getBytes()); + connection.setCommands().sRem(key.getBytes(), values); } } - @Override - public void evict(String key) { + private Set processGetMembers(String key) { try (RedisConnection connection = redisConnectionFactory.getConnection()) { - connection.del(key.getBytes()); + 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 index 29140aea30..3f6697e219 100644 --- 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 @@ -15,16 +15,31 @@ */ 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, byte[]... values); + void add(String key, String value); - void remove(String key, byte[]... values); + void add(String key, EntityId value); - Set get(String key); + void add(String key, TbMsg value); - void evict(String key); + 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/application/src/main/java/org/thingsboard/server/service/rule/RuleNodeCacheService.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleNodeCacheService.java similarity index 52% rename from application/src/main/java/org/thingsboard/server/service/rule/RuleNodeCacheService.java rename to rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleNodeCacheService.java index 4ccbb3b6ef..d627923168 100644 --- a/application/src/main/java/org/thingsboard/server/service/rule/RuleNodeCacheService.java +++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleNodeCacheService.java @@ -13,9 +13,9 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.thingsboard.server.service.rule; +package org.thingsboard.rule.engine.api; -import org.thingsboard.server.common.data.id.RuleNodeId; +import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.msg.TbMsg; import java.util.List; @@ -23,21 +23,24 @@ import java.util.Set; public interface RuleNodeCacheService { - void add(RuleNodeId ruleNodeId, String key, byte[]... values); + void add(String key, String value); - void add(RuleNodeId ruleNodeId, String key, TbMsg value); + void add(String key, EntityId id); - void add(RuleNodeId ruleNodeId, String key, List values); + void add(String key, TbMsg value); - Set get(RuleNodeId ruleNodeId, String key); + void removeStringList(String key, List values); - Set get(RuleNodeId ruleNodeId, String key, String queueName); + void removeEntityIdList(String key, List values); - void remove(RuleNodeId ruleNodeId, String key, byte[]... values); + void removeTbMsgList(String key, List values); - void remove(RuleNodeId ruleNodeId, String key, TbMsg value); + Set getStrings(String key); - void remove(RuleNodeId ruleNodeId, String key, List values); + Set getEntityIds(String key); + + Set getTbMsgs(String key, String queueName); + + void evict(String key); - void evict(RuleNodeId ruleNodeId, 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 3456c23f3c..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,27 +317,7 @@ public interface TbContext { void clearRuleNodeStates(); - void addToRuleNodeCache(String key, TbMsg msg); - - void addToRuleNodeCache(String key, List tbMsgList); - - void addToRuleNodeCache(String key, String value); - - void addToRuleNodeCache(String key, byte[]... values); - - Set getFromRuleNodeCache(String key); - - Set getFromRuleNodeCache(String key, String queueName); - - void removeFromRuleNodeCache(String key, TbMsg msg); - - void removeFromRuleNodeCache(String key, List tbMsgList); - - void removeFromRuleNodeCache(String key, byte[]... values); - - void removeFromRuleNodeCache(String key, String value); - - void evictFromRuleNodeCache(String key); + RuleNodeCacheService getRuleNodeCacheService(); void addTenantProfileListener(Consumer listener); 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 34083d252b..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,8 +31,11 @@ public interface TbNode { default void destroy() {} - default void onPartitionChangeMsg(TbContext ctx, PartitionChangeMsg msg) {} + // todo: add reason for destroy + default void destroy(TbContext ctx) { + destroy(); + } - default void onRemove(TbContext ctx) {} + 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/TbMsgDeduplicationNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/deduplication/TbMsgDeduplicationNode.java index 7dc588cb97..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 @@ -18,7 +18,6 @@ package org.thingsboard.rule.engine.deduplication; import com.fasterxml.jackson.databind.node.ArrayNode; import com.fasterxml.jackson.databind.node.ObjectNode; import lombok.extern.slf4j.Slf4j; -import org.springframework.util.SerializationUtils; import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.rule.engine.api.RuleNode; import org.thingsboard.rule.engine.api.TbContext; @@ -43,7 +42,6 @@ import java.util.Optional; import java.util.Set; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; -import java.util.stream.Collectors; @RuleNode( type = ComponentType.ACTION, @@ -97,14 +95,14 @@ public class TbMsgDeduplicationNode implements TbNode { } @Override - public void onRemove(TbContext ctx) { + public void destroy(TbContext ctx) { deduplicationMap.clear(); Set deduplicationIds = getDeduplicationIds(ctx); if (deduplicationIds == null) { return; } - deduplicationIds.forEach(id -> ctx.evictFromRuleNodeCache(id.toString())); - ctx.evictFromRuleNodeCache(DEDUPLICATION_IDS_CACHE_KEY); + deduplicationIds.forEach(id -> ctx.getRuleNodeCacheService().evict(id.toString())); + ctx.getRuleNodeCacheService().evict(DEDUPLICATION_IDS_CACHE_KEY); } private void getDeduplicationDataFromCacheAndSchedule(TbContext ctx) { @@ -113,7 +111,7 @@ public class TbMsgDeduplicationNode implements TbNode { return; } deduplicationIds.forEach(id -> { - List tbMsgs = new ArrayList<>(ctx.getFromRuleNodeCache(id.toString(), config.getQueueName())); + List tbMsgs = new ArrayList<>(ctx.getRuleNodeCacheService().getTbMsgs(id.toString(), config.getQueueName())); DeduplicationData deduplicationData = new DeduplicationData(); deduplicationData.addAll(tbMsgs); deduplicationMap.put(id, deduplicationData); @@ -122,10 +120,7 @@ public class TbMsgDeduplicationNode implements TbNode { } private Set getDeduplicationIds(TbContext ctx) { - Set deduplicationIds = ctx.getFromRuleNodeCache(TbMsgDeduplicationNode.DEDUPLICATION_IDS_CACHE_KEY) - .stream() - .map(bytes -> (EntityId) SerializationUtils.deserialize(bytes)) - .collect(Collectors.toSet()); + Set deduplicationIds = ctx.getRuleNodeCacheService().getEntityIds(TbMsgDeduplicationNode.DEDUPLICATION_IDS_CACHE_KEY); if (deduplicationIds.isEmpty()) { return null; } @@ -138,10 +133,10 @@ public class TbMsgDeduplicationNode implements TbNode { 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.addToRuleNodeCache(DEDUPLICATION_IDS_CACHE_KEY, SerializationUtils.serialize(id)); + ctx.getRuleNodeCacheService().add(DEDUPLICATION_IDS_CACHE_KEY, id); } deduplicationMsgs.add(msg); - ctx.addToRuleNodeCache(id.toString(), msg); + ctx.getRuleNodeCacheService().add(id.toString(), msg); ctx.ack(msg); scheduleTickMsg(ctx, id, deduplicationMsgs); } else { @@ -252,7 +247,7 @@ public class TbMsgDeduplicationNode implements TbNode { ctx.enqueueForTellNext(outMsg, TbRelationTypes.SUCCESS, () -> { log.trace("[{}][{}][{}] Successfully enqueue deduplication result message!", ctx.getSelfId(), outMsg.getOriginator(), retryAttempt); - ctx.removeFromRuleNodeCache(outMsg.getOriginator().toString(), msgsToRemoveFromCache); + ctx.getRuleNodeCacheService().removeTbMsgList(outMsg.getOriginator().toString(), msgsToRemoveFromCache); }, throwable -> { log.trace("[{}][{}][{}] Failed to enqueue deduplication output message due to: ", ctx.getSelfId(), outMsg.getOriginator(), retryAttempt, throwable); diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbMsgDeduplicationNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbMsgDeduplicationNodeTest.java index 299e9f896d..56a5f4b707 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbMsgDeduplicationNodeTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbMsgDeduplicationNodeTest.java @@ -28,6 +28,7 @@ import org.mockito.ArgumentMatchers; import org.mockito.stubbing.Answer; import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.common.util.ThingsBoardThreadFactory; +import org.thingsboard.rule.engine.api.RuleNodeCacheService; import org.thingsboard.rule.engine.api.TbContext; import org.thingsboard.rule.engine.api.TbNodeConfiguration; import org.thingsboard.rule.engine.api.TbNodeException; @@ -44,6 +45,7 @@ import org.thingsboard.server.common.msg.TbMsgMetaData; import org.thingsboard.server.common.msg.session.SessionMsgType; import java.util.ArrayList; +import java.util.Collections; import java.util.List; import java.util.Random; import java.util.UUID; @@ -58,11 +60,15 @@ import java.util.function.Consumer; import java.util.stream.Collectors; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyList; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.ArgumentMatchers.isNull; import static org.mockito.ArgumentMatchers.nullable; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.spy; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; @@ -74,6 +80,7 @@ public class TbMsgDeduplicationNodeTest { private static final String MAIN_QUEUE_NAME = "Main"; private static final String HIGH_PRIORITY_QUEUE_NAME = "HighPriority"; private static final String TB_MSG_DEDUPLICATION_TIMEOUT_MSG = "TbMsgDeduplicationNodeMsg"; + private static final String DEDUPLICATION_IDS_CACHE_KEY = "deduplication_ids"; private TbContext ctx; @@ -86,18 +93,24 @@ public class TbMsgDeduplicationNodeTest { private TbMsgDeduplicationNode node; private TbMsgDeduplicationNodeConfiguration config; private TbNodeConfiguration nodeConfiguration; + private RuleNodeCacheService ruleNodeCacheService; private CountDownLatch awaitTellSelfLatch; @BeforeEach public void init() throws TbNodeException { ctx = mock(TbContext.class); + ruleNodeCacheService = mock(RuleNodeCacheService.class); tenantId = TenantId.fromUUID(UUID.randomUUID()); RuleNodeId ruleNodeId = new RuleNodeId(UUID.randomUUID()); when(ctx.getSelfId()).thenReturn(ruleNodeId); when(ctx.getTenantId()).thenReturn(tenantId); + when(ctx.getRuleNodeCacheService()).thenReturn(ruleNodeCacheService); + when(ruleNodeCacheService.getStrings(anyString())).thenReturn(Collections.emptySet()); + when(ruleNodeCacheService.getEntityIds(anyString())).thenReturn(Collections.emptySet()); + when(ruleNodeCacheService.getTbMsgs(anyString(), anyString())).thenReturn(Collections.emptySet()); doAnswer((Answer) invocationOnMock -> { String type = (String) (invocationOnMock.getArguments())[1]; @@ -173,10 +186,22 @@ 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()); + + 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())); } @@ -212,10 +237,22 @@ 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()); + + 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())); } @@ -247,10 +284,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()); @@ -293,10 +341,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()); @@ -347,10 +406,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()); List resultMsgsByteStrings = resultMsgs.stream().map(TbMsg::toByteString).collect(Collectors.toList());