From 329eb29f4134a8be627fc9498b83e55d1038dc64 Mon Sep 17 00:00:00 2001 From: ShvaykaD Date: Tue, 24 Jan 2023 12:27:20 +0200 Subject: [PATCH 1/4] added ability to store msgs in cache --- .../server/actors/ActorSystemContext.java | 5 + .../actors/ruleChain/DefaultTbContext.java | 55 +++++++++++ .../RuleNodeActorMessageProcessor.java | 1 + .../rule/DefaultRuleNodeCacheService.java | 95 +++++++++++++++++++ .../service/rule/RuleNodeCacheService.java | 43 +++++++++ .../cache/rule/DefaultRuleNodeCache.java | 64 +++++++++++++ .../server/cache/rule/RedisRuleNodeCache.java | 61 ++++++++++++ .../server/cache/rule/RuleNodeCache.java | 30 ++++++ .../rule/engine/api/TbContext.java | 22 +++++ .../thingsboard/rule/engine/api/TbNode.java | 2 + .../deduplication/DeduplicationData.java | 4 + .../deduplication/TbMsgDeduplicationNode.java | 81 ++++++++++++---- .../action/TbMsgDeduplicationNodeTest.java | 4 +- 13 files changed, 448 insertions(+), 19 deletions(-) create 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/RuleNodeCacheService.java create mode 100644 common/cache/src/main/java/org/thingsboard/server/cache/rule/DefaultRuleNodeCache.java create mode 100644 common/cache/src/main/java/org/thingsboard/server/cache/rule/RedisRuleNodeCache.java create mode 100644 common/cache/src/main/java/org/thingsboard/server/cache/rule/RuleNodeCache.java 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..511211c303 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java +++ b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java @@ -96,6 +96,7 @@ 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; @@ -234,6 +235,10 @@ public class ActorSystemContext { @Getter private RuleNodeStateService ruleNodeStateService; + @Autowired + @Getter + private RuleNodeCacheService ruleNodeCacheService; + @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..398d9c3a8a 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 @@ -733,6 +733,61 @@ class DefaultTbContext implements TbContext { mainCtx.getRuleNodeStateService().removeByRuleNodeId(getTenantId(), getSelfId()); } + @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); + } + @Override public void removeRuleNodeStateForEntity(EntityId entityId) { if (log.isDebugEnabled()) { 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..06c5e9e7e1 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 @@ -89,6 +89,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/RuleNodeCacheService.java b/application/src/main/java/org/thingsboard/server/service/rule/RuleNodeCacheService.java new file mode 100644 index 0000000000..4ccbb3b6ef --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/rule/RuleNodeCacheService.java @@ -0,0 +1,43 @@ +/** + * 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.server.common.data.id.RuleNodeId; +import org.thingsboard.server.common.msg.TbMsg; + +import java.util.List; +import java.util.Set; + +public interface RuleNodeCacheService { + + void add(RuleNodeId ruleNodeId, String key, byte[]... values); + + void add(RuleNodeId ruleNodeId, String key, TbMsg value); + + void add(RuleNodeId ruleNodeId, String key, List values); + + Set get(RuleNodeId ruleNodeId, String key); + + Set get(RuleNodeId ruleNodeId, String key, String queueName); + + void remove(RuleNodeId ruleNodeId, String key, byte[]... values); + + void remove(RuleNodeId ruleNodeId, String key, TbMsg value); + + void remove(RuleNodeId ruleNodeId, String key, List values); + + void evict(RuleNodeId ruleNodeId, String key); +} 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..cc76cf9688 --- /dev/null +++ b/common/cache/src/main/java/org/thingsboard/server/cache/rule/DefaultRuleNodeCache.java @@ -0,0 +1,64 @@ +/** + * 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.extern.slf4j.Slf4j; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.stereotype.Service; + +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.Set; +import java.util.TreeSet; + +@Service +@ConditionalOnProperty(prefix = "cache", value = "type", havingValue = "caffeine", matchIfMissing = true) +@Slf4j +public class DefaultRuleNodeCache implements RuleNodeCache { + + private final Map> cache; + + public DefaultRuleNodeCache() { + cache = new HashMap<>(); + } + + @Override + public void add(String key, byte[]... values) { + Set data = cache.computeIfAbsent(key, s -> new TreeSet<>(Arrays::compare)); + data.addAll(Arrays.asList(values)); + } + + @Override + public void remove(String key, byte[]... values) { + cache.computeIfPresent(key, (id, data) -> { + Arrays.asList(values).forEach(data::remove); + return data; + }); + } + + @Override + public Set get(String key) { + return cache.getOrDefault(key, 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 new file mode 100644 index 0000000000..bc583074b0 --- /dev/null +++ b/common/cache/src/main/java/org/thingsboard/server/cache/rule/RedisRuleNodeCache.java @@ -0,0 +1,61 @@ +/** + * 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 java.util.Set; + +@Service +@ConditionalOnProperty(prefix = "cache", value = "type", havingValue = "redis") +@RequiredArgsConstructor +public class RedisRuleNodeCache implements RuleNodeCache { + + private final RedisConnectionFactory redisConnectionFactory; + + @Override + public void add(String key, byte[]... values) { + try (RedisConnection connection = redisConnectionFactory.getConnection()) { + connection.setCommands().sAdd(key.getBytes(), values); + } + } + + @Override + public void remove(String key, byte[]... values) { + try (RedisConnection connection = redisConnectionFactory.getConnection()) { + connection.setCommands().sRem(key.getBytes(), values); + } + } + + @Override + public Set get(String key) { + try (RedisConnection connection = redisConnectionFactory.getConnection()) { + return connection.setCommands().sMembers(key.getBytes()); + } + } + + @Override + public void evict(String key) { + try (RedisConnection connection = redisConnectionFactory.getConnection()) { + connection.del(key.getBytes()); + } + } + +} 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..29140aea30 --- /dev/null +++ b/common/cache/src/main/java/org/thingsboard/server/cache/rule/RuleNodeCache.java @@ -0,0 +1,30 @@ +/** + * 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 java.util.Set; + +public interface RuleNodeCache { + + void add(String key, byte[]... values); + + void remove(String key, byte[]... values); + + Set get(String key); + + 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..3456c23f3c 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,28 @@ 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); + 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..34083d252b 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 @@ -33,4 +33,6 @@ public interface TbNode { default void onPartitionChangeMsg(TbContext ctx, PartitionChangeMsg msg) {} + default void onRemove(TbContext ctx) {} + } 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..7dc588cb97 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,6 +18,7 @@ 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; @@ -39,8 +40,10 @@ 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; +import java.util.stream.Collectors; @RuleNode( type = ComponentType.ACTION, @@ -59,6 +62,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 +82,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 +97,39 @@ public class TbMsgDeduplicationNode implements TbNode { } @Override - public void destroy() { + public void onRemove(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); + } + + private void getDeduplicationDataFromCacheAndSchedule(TbContext ctx) { + Set deduplicationIds = getDeduplicationIds(ctx); + if (deduplicationIds == null) { + return; + } + deduplicationIds.forEach(id -> { + List tbMsgs = new ArrayList<>(ctx.getFromRuleNodeCache(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.getFromRuleNodeCache(TbMsgDeduplicationNode.DEDUPLICATION_IDS_CACHE_KEY) + .stream() + .map(bytes -> (EntityId) SerializationUtils.deserialize(bytes)) + .collect(Collectors.toSet()); + if (deduplicationIds.isEmpty()) { + return null; + } + return deduplicationIds; } private void processOnRegularMsg(TbContext ctx, TbMsg msg) { @@ -99,7 +137,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.addToRuleNodeCache(DEDUPLICATION_IDS_CACHE_KEY, SerializationUtils.serialize(id)); + } deduplicationMsgs.add(msg); + ctx.addToRuleNodeCache(id.toString(), msg); ctx.ack(msg); scheduleTickMsg(ctx, id, deduplicationMsgs); } else { @@ -132,13 +174,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 +189,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 +202,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 +211,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 +245,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.removeFromRuleNodeCache(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 +276,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/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 923d5667f4..2dc4054c50 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 @@ -75,7 +75,7 @@ public class TbMsgDeduplicationNodeTest { private TbContext ctx; - private final ThingsBoardThreadFactory factory = ThingsBoardThreadFactory.forName("de-duplication-node-test"); + private final ThingsBoardThreadFactory factory = ThingsBoardThreadFactory.forName("deduplication-node-test"); private final ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor(factory); private final int deduplicationInterval = 1; @@ -278,7 +278,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) { From 97f01314f4b8c7deaf5b4f72fa6786f3eb0fd388 Mon Sep 17 00:00:00 2001 From: ShvaykaD Date: Tue, 24 Jan 2023 12:45:05 +0200 Subject: [PATCH 2/4] updated tests and default rule node configuration --- .../server/cache/rule/DefaultRuleNodeCache.java | 1 - .../TbMsgDeduplicationNodeConfiguration.java | 3 +++ .../engine/action/TbMsgDeduplicationNodeTest.java | 11 +++++++---- 3 files changed, 10 insertions(+), 5 deletions(-) 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 cc76cf9688..9631df5e08 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 @@ -28,7 +28,6 @@ import java.util.TreeSet; @Service @ConditionalOnProperty(prefix = "cache", value = "type", havingValue = "caffeine", matchIfMissing = true) -@Slf4j public class DefaultRuleNodeCache implements RuleNodeCache { private final Map> cache; 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..8f3fa6f371 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 @@ -21,6 +21,8 @@ import org.thingsboard.rule.engine.api.NodeConfiguration; @Data public class TbMsgDeduplicationNodeConfiguration implements NodeConfiguration { + private static final String DEFAULT_QUEUE_NAME = "Main"; + private int interval; private DeduplicationId id; private DeduplicationStrategy strategy; @@ -39,6 +41,7 @@ public class TbMsgDeduplicationNodeConfiguration implements NodeConfiguration 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) { From 2f26b8dbc0c01465152a2e343150bff07a2f3148 Mon Sep 17 00:00:00 2001 From: ShvaykaD Date: Tue, 24 Jan 2023 17:56:32 +0200 Subject: [PATCH 3/4] 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()); From 5c40104be082e8eb272ffb002184f35741e78e61 Mon Sep 17 00:00:00 2001 From: ShvaykaD Date: Tue, 24 Jan 2023 18:21:55 +0200 Subject: [PATCH 4/4] move default queue names and topics to DataConstants --- .../DefaultSystemDataLoaderService.java | 22 +++++++++---------- .../TbRuleEngineQueueConfigService.java | 3 ++- .../update/DefaultDataUpdateService.java | 5 +++-- .../controller/BaseTenantControllerTest.java | 5 +++-- .../BaseTenantProfileControllerTest.java | 5 +++-- .../server/edge/AbstractEdgeTest.java | 5 +++-- .../queue/DefaultTbClusterServiceTest.java | 3 ++- .../queue/TbMsgPackProcessingContextTest.java | 3 ++- .../server/common/data/DataConstants.java | 8 +++++++ .../server/queue/discovery/QueueKey.java | 6 ++--- .../server/queue/discovery/QueueKeyTest.java | 5 +++-- .../validator/TenantProfileDataValidator.java | 3 ++- .../server/dao/sql/queue/JpaQueueDao.java | 3 ++- .../dao/service/BaseQueueServiceTest.java | 7 +++--- .../service/BaseTenantProfileServiceTest.java | 9 ++++---- .../TbMsgDeduplicationNodeConfiguration.java | 7 +++--- .../TbMsgDeduplicationNodeTest.java | 11 +++++----- 17 files changed, 64 insertions(+), 46 deletions(-) rename rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/{action => transform}/TbMsgDeduplicationNodeTest.java (98%) diff --git a/application/src/main/java/org/thingsboard/server/service/install/DefaultSystemDataLoaderService.java b/application/src/main/java/org/thingsboard/server/service/install/DefaultSystemDataLoaderService.java index 9507912c39..1252c50e00 100644 --- a/application/src/main/java/org/thingsboard/server/service/install/DefaultSystemDataLoaderService.java +++ b/application/src/main/java/org/thingsboard/server/service/install/DefaultSystemDataLoaderService.java @@ -203,8 +203,8 @@ public class DefaultSystemDataLoaderService implements SystemDataLoaderService { isolatedRuleEngineTenantProfileData.setConfiguration(new DefaultTenantProfileConfiguration()); 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); @@ -598,12 +598,12 @@ public class DefaultSystemDataLoaderService implements SystemDataLoaderService { @Override public void createQueues() { - Queue mainQueue = queueService.findQueueByTenantIdAndName(TenantId.SYS_TENANT_ID, "Main"); + Queue mainQueue = queueService.findQueueByTenantIdAndName(TenantId.SYS_TENANT_ID, DataConstants.MAIN_QUEUE_NAME); if (mainQueue == null) { mainQueue = new Queue(); mainQueue.setTenantId(TenantId.SYS_TENANT_ID); - mainQueue.setName("Main"); - mainQueue.setTopic("tb_rule_engine.main"); + mainQueue.setName(DataConstants.MAIN_QUEUE_NAME); + mainQueue.setTopic(DataConstants.MAIN_QUEUE_TOPIC); mainQueue.setPollInterval(25); mainQueue.setPartitions(10); mainQueue.setConsumerPerPartition(true); @@ -622,12 +622,12 @@ public class DefaultSystemDataLoaderService implements SystemDataLoaderService { queueService.saveQueue(mainQueue); } - Queue highPriorityQueue = queueService.findQueueByTenantIdAndName(TenantId.SYS_TENANT_ID, "HighPriority"); + Queue highPriorityQueue = queueService.findQueueByTenantIdAndName(TenantId.SYS_TENANT_ID, DataConstants.HP_QUEUE_NAME); if (highPriorityQueue == null) { highPriorityQueue = new Queue(); highPriorityQueue.setTenantId(TenantId.SYS_TENANT_ID); - highPriorityQueue.setName("HighPriority"); - highPriorityQueue.setTopic("tb_rule_engine.hp"); + highPriorityQueue.setName(DataConstants.HP_QUEUE_NAME); + highPriorityQueue.setTopic(DataConstants.HP_QUEUE_TOPIC); highPriorityQueue.setPollInterval(25); highPriorityQueue.setPartitions(10); highPriorityQueue.setConsumerPerPartition(true); @@ -646,12 +646,12 @@ public class DefaultSystemDataLoaderService implements SystemDataLoaderService { queueService.saveQueue(highPriorityQueue); } - Queue sequentialByOriginatorQueue = queueService.findQueueByTenantIdAndName(TenantId.SYS_TENANT_ID, "SequentialByOriginator"); + Queue sequentialByOriginatorQueue = queueService.findQueueByTenantIdAndName(TenantId.SYS_TENANT_ID, DataConstants.SQ_QUEUE_NAME); if (sequentialByOriginatorQueue == null) { sequentialByOriginatorQueue = new Queue(); sequentialByOriginatorQueue.setTenantId(TenantId.SYS_TENANT_ID); - sequentialByOriginatorQueue.setName("SequentialByOriginator"); - sequentialByOriginatorQueue.setTopic("tb_rule_engine.sq"); + sequentialByOriginatorQueue.setName(DataConstants.SQ_QUEUE_NAME); + sequentialByOriginatorQueue.setTopic(DataConstants.SQ_QUEUE_TOPIC); sequentialByOriginatorQueue.setPollInterval(25); sequentialByOriginatorQueue.setPartitions(10); sequentialByOriginatorQueue.setPackProcessingTimeout(2000); diff --git a/application/src/main/java/org/thingsboard/server/service/install/TbRuleEngineQueueConfigService.java b/application/src/main/java/org/thingsboard/server/service/install/TbRuleEngineQueueConfigService.java index 1125467082..53bd86f810 100644 --- a/application/src/main/java/org/thingsboard/server/service/install/TbRuleEngineQueueConfigService.java +++ b/application/src/main/java/org/thingsboard/server/service/install/TbRuleEngineQueueConfigService.java @@ -21,6 +21,7 @@ import org.springframework.boot.autoconfigure.EnableAutoConfiguration; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Profile; +import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration; import javax.annotation.PostConstruct; @@ -39,7 +40,7 @@ public class TbRuleEngineQueueConfigService { @PostConstruct public void validate() { - queues.stream().filter(queue -> 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/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/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-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 8f3fa6f371..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,12 +17,11 @@ 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 { - private static final String DEFAULT_QUEUE_NAME = "Main"; - private int interval; private DeduplicationId id; private DeduplicationStrategy strategy; @@ -30,10 +29,10 @@ public class TbMsgDeduplicationNodeConfiguration implements NodeConfiguration