From 85d229be2d15e6bff0c4249f8f6ccd32a88c0356 Mon Sep 17 00:00:00 2001 From: ShvaykaD Date: Fri, 29 Mar 2024 12:05:44 +0200 Subject: [PATCH 1/5] convert ValueWithTs to a record & replace usages of Collections.singletonList to List.of & use single key search for async method --- .../dao/timeseries/TimeseriesService.java | 2 +- .../engine/metadata/CalculateDeltaNode.java | 34 ++++++------------- .../metadata/CalculateDeltaNodeTest.java | 11 +++--- 3 files changed, 17 insertions(+), 30 deletions(-) diff --git a/common/dao-api/src/main/java/org/thingsboard/server/dao/timeseries/TimeseriesService.java b/common/dao-api/src/main/java/org/thingsboard/server/dao/timeseries/TimeseriesService.java index 096a39bff6..c62898ab09 100644 --- a/common/dao-api/src/main/java/org/thingsboard/server/dao/timeseries/TimeseriesService.java +++ b/common/dao-api/src/main/java/org/thingsboard/server/dao/timeseries/TimeseriesService.java @@ -38,7 +38,7 @@ public interface TimeseriesService { ListenableFuture> findAll(TenantId tenantId, EntityId entityId, List queries); - ListenableFuture> findLatest(TenantId tenantId, EntityId entityId, String keys); + ListenableFuture> findLatest(TenantId tenantId, EntityId entityId, String key); ListenableFuture> findLatest(TenantId tenantId, EntityId entityId, Collection keys); diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNode.java index 609888db84..88540ffe4a 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNode.java @@ -37,7 +37,6 @@ import org.thingsboard.server.dao.timeseries.TimeseriesService; import java.math.BigDecimal; import java.math.RoundingMode; -import java.util.Collections; import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; @@ -129,8 +128,8 @@ public class CalculateDeltaNode implements TbNode { } private ListenableFuture fetchLatestValueAsync(EntityId entityId) { - return Futures.transform(timeseriesService.findLatest(ctx.getTenantId(), entityId, Collections.singletonList(config.getInputValueKey())), - list -> extractValue(list.get(0)) + return Futures.transform(timeseriesService.findLatest(ctx.getTenantId(), entityId, config.getInputValueKey()), + tsKvEntryOpt -> tsKvEntryOpt.map(this::extractValue).orElse(null) , ctx.getDbCallbackExecutor()); } @@ -138,7 +137,7 @@ public class CalculateDeltaNode implements TbNode { List tsKvEntries = timeseriesService.findLatestSync( ctx.getTenantId(), entityId, - Collections.singletonList(config.getInputValueKey())); + List.of(config.getInputValueKey())); return extractValue(tsKvEntries.get(0)); } @@ -161,36 +160,23 @@ public class CalculateDeltaNode implements TbNode { double result = 0.0; long ts = kvEntry.getTs(); switch (kvEntry.getDataType()) { - case LONG: - result = kvEntry.getLongValue().get(); - break; - case DOUBLE: - result = kvEntry.getDoubleValue().get(); - break; - case STRING: + case LONG -> result = kvEntry.getLongValue().get(); + case DOUBLE -> result = kvEntry.getDoubleValue().get(); + case STRING -> { try { result = Double.parseDouble(kvEntry.getStrValue().get()); } catch (NumberFormatException e) { throw new IllegalArgumentException("Calculation failed. Unable to parse value [" + kvEntry.getStrValue().get() + "]" + " of telemetry [" + kvEntry.getKey() + "] to Double"); } - break; - case BOOLEAN: - throw new IllegalArgumentException("Calculation failed. Boolean values are not supported!"); - case JSON: - throw new IllegalArgumentException("Calculation failed. JSON values are not supported!"); + } + case BOOLEAN -> throw new IllegalArgumentException("Calculation failed. Boolean values are not supported!"); + case JSON -> throw new IllegalArgumentException("Calculation failed. JSON values are not supported!"); } return new ValueWithTs(ts, result); } - private static class ValueWithTs { - private final long ts; - private final double value; - - private ValueWithTs(long ts, double value) { - this.ts = ts; - this.value = value; - } + private record ValueWithTs(long ts, double value) { } } diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNodeTest.java index 707de1741f..188dd2e4ec 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNodeTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNodeTest.java @@ -47,6 +47,7 @@ import org.thingsboard.server.common.msg.TbMsgMetaData; import org.thingsboard.server.dao.timeseries.TimeseriesService; import java.util.List; +import java.util.Optional; import java.util.UUID; import static org.junit.jupiter.api.Assertions.assertEquals; @@ -68,9 +69,9 @@ import static org.mockito.Mockito.when; @ExtendWith(MockitoExtension.class) public class CalculateDeltaNodeTest { - private static final DeviceId DUMMY_DEVICE_ORIGINATOR = new DeviceId(UUID.randomUUID()); - private static final TenantId TENANT_ID = new TenantId(UUID.randomUUID()); - private static final ListeningExecutor DB_EXECUTOR = new TestDbCallbackExecutor(); + private final DeviceId DUMMY_DEVICE_ORIGINATOR = new DeviceId(UUID.fromString("2ba3ded4-882b-40cf-999a-89da9ccd58f9")); + private final TenantId TENANT_ID = new TenantId(UUID.fromString("3842e740-0d89-43a9-8d52-ae44023847ba")); + private final ListeningExecutor DB_EXECUTOR = new TestDbCallbackExecutor(); @Mock private TbContext ctxMock; @Mock @@ -435,8 +436,8 @@ public class CalculateDeltaNodeTest { when(ctxMock.getDbCallbackExecutor()).thenReturn(DB_EXECUTOR); when(ctxMock.getTenantId()).thenReturn(TENANT_ID); when(timeseriesServiceMock.findLatest( - eq(TENANT_ID), eq(DUMMY_DEVICE_ORIGINATOR), argThat(new ListMatcher<>(List.of(tsKvEntry.getKey()))) - )).thenReturn(Futures.immediateFuture(List.of(tsKvEntry))); + eq(TENANT_ID), eq(DUMMY_DEVICE_ORIGINATOR), eq(tsKvEntry.getKey()) + )).thenReturn(Futures.immediateFuture(Optional.of(tsKvEntry))); } @RequiredArgsConstructor From 95fad7bf1bfe35274cf5dd8628278ea9c2148b55 Mon Sep 17 00:00:00 2001 From: ShvaykaD Date: Mon, 1 Apr 2024 12:36:49 +0300 Subject: [PATCH 2/5] moved SemaphoreWithTbMsgQueue to separate file & updated logic in CalculateDeltaNode --- .../rule/engine/math/TbMathNode.java | 91 +----------- .../engine/metadata/CalculateDeltaNode.java | 119 +++++++-------- .../engine/util/SemaphoreWithTbMsgQueue.java | 140 ++++++++++++++++++ .../rule/engine/math/TbMathNodeTest.java | 14 +- .../metadata/CalculateDeltaNodeTest.java | 110 +++++++------- 5 files changed, 266 insertions(+), 208 deletions(-) create mode 100644 rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/util/SemaphoreWithTbMsgQueue.java diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/math/TbMathNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/math/TbMathNode.java index 3813195e23..823c224833 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/math/TbMathNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/math/TbMathNode.java @@ -19,13 +19,10 @@ import com.fasterxml.jackson.databind.node.ObjectNode; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.MoreExecutors; -import lombok.Data; -import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import net.objecthunter.exp4j.Expression; import net.objecthunter.exp4j.ExpressionBuilder; import org.springframework.util.ConcurrentReferenceHashMap; -import org.thingsboard.common.util.DonAsynchron; import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.rule.engine.api.RuleNode; import org.thingsboard.rule.engine.api.TbContext; @@ -33,6 +30,7 @@ import org.thingsboard.rule.engine.api.TbNode; import org.thingsboard.rule.engine.api.TbNodeConfiguration; import org.thingsboard.rule.engine.api.TbNodeException; import org.thingsboard.rule.engine.api.util.TbNodeUtils; +import org.thingsboard.rule.engine.util.SemaphoreWithTbMsgQueue; import org.thingsboard.server.common.data.AttributeScope; import org.thingsboard.server.common.data.StringUtils; import org.thingsboard.server.common.data.id.EntityId; @@ -46,10 +44,7 @@ import java.math.BigDecimal; import java.math.RoundingMode; import java.util.List; import java.util.Optional; -import java.util.Queue; -import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.ConcurrentMap; -import java.util.concurrent.Semaphore; import java.util.function.BiFunction; import java.util.function.Function; import java.util.stream.Collectors; @@ -84,7 +79,7 @@ import static org.thingsboard.rule.engine.math.TbMathArgumentType.CONSTANT; ) public class TbMathNode implements TbNode { - private static final ConcurrentMap> locks = new ConcurrentReferenceHashMap<>(16, ConcurrentReferenceHashMap.ReferenceType.WEAK); + private static final ConcurrentMap locks = new ConcurrentReferenceHashMap<>(16, ConcurrentReferenceHashMap.ReferenceType.WEAK); private final ThreadLocal customExpression = new ThreadLocal<>(); private TbMathNodeConfiguration config; private boolean msgBodyToJsonConversionRequired; @@ -110,66 +105,8 @@ public class TbMathNode implements TbNode { @Override public void onMsg(TbContext ctx, TbMsg msg) { - var semaphoreWithQueue = locks.computeIfAbsent(msg.getOriginator(), SemaphoreWithQueue::new); - semaphoreWithQueue.getQueue().add(new TbMsgTbContextBiFunction(msg, ctx, this::processMsgAsync)); - - tryProcessQueue(semaphoreWithQueue); - } - - void tryProcessQueue(SemaphoreWithQueue lockAndQueue) { - final Semaphore semaphore = lockAndQueue.getSemaphore(); - final Queue queue = lockAndQueue.getQueue(); - while (!queue.isEmpty()) { - // The semaphore have to be acquired before EACH poll and released before NEXT poll. - // Otherwise, some message will remain unprocessed in queue - if (!semaphore.tryAcquire()) { - return; - } - TbMsgTbContextBiFunction tbMsgTbContext = null; - try { - tbMsgTbContext = queue.poll(); - if (tbMsgTbContext == null) { - semaphore.release(); - continue; - } - final TbMsg msg = tbMsgTbContext.getMsg(); - if (!msg.getCallback().isMsgValid()) { - log.trace("[{}] Skipping non-valid message [{}]", lockAndQueue.getEntityId(), msg); - semaphore.release(); - continue; - } - //DO PROCESSING - final TbContext ctx = tbMsgTbContext.getCtx(); - final ListenableFuture resultMsgFuture = tbMsgTbContext.getBiFunction().apply(ctx, msg); - DonAsynchron.withCallback(resultMsgFuture, resultMsg -> { - try { - ctx.tellSuccess(resultMsg); - } finally { - lockAndQueue.getSemaphore().release(); - tryProcessQueue(lockAndQueue); - } - }, t -> { - try { - ctx.tellFailure(msg, t); - } finally { - lockAndQueue.getSemaphore().release(); - tryProcessQueue(lockAndQueue); - } - }, ctx.getDbCallbackExecutor()); - } catch (Throwable t) { - semaphore.release(); - if (tbMsgTbContext == null) { // if no message polled, the loop become infinite, will throw exception - log.error("[{}] Failed to process TbMsgTbContext queue", lockAndQueue.getEntityId(), t); - throw t; - } - TbMsg msg = tbMsgTbContext.getMsg(); - TbContext ctx = tbMsgTbContext.getCtx(); - log.warn("[{}] Failed to process message: {}", lockAndQueue.getEntityId(), msg, t); - ctx.tellFailure(msg, t); // you are not allowed to throw here, because queue will remain unprocessed - continue; // We are probably the last who process the queue. We have to continue poll until get successful callback or queue is empty - } - break; //submitted async exact one task. next poll will try on callback - } + var semaphoreWithQueue = locks.computeIfAbsent(msg.getOriginator(), SemaphoreWithTbMsgQueue::new); + semaphoreWithQueue.addToQueueAndTryProcess(msg, ctx, this::processMsgAsync); } ListenableFuture processMsgAsync(TbContext ctx, TbMsg msg) { @@ -426,24 +363,4 @@ public class TbMathNode implements TbNode { } } - @Override - public void destroy() { - } - - @Data - @RequiredArgsConstructor - static public class SemaphoreWithQueue { - final EntityId entityId; - final Semaphore semaphore = new Semaphore(1); - final Queue queue = new ConcurrentLinkedQueue<>(); - } - - @Data - @RequiredArgsConstructor - static public class TbMsgTbContextBiFunction { - final TbMsg msg; - final TbContext ctx; - final BiFunction> biFunction; - } - } diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNode.java index 88540ffe4a..f459a532ed 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNode.java @@ -19,7 +19,9 @@ import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.node.ObjectNode; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; +import com.google.common.util.concurrent.MoreExecutors; import lombok.extern.slf4j.Slf4j; +import org.springframework.util.ConcurrentReferenceHashMap; import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.rule.engine.api.RuleNode; import org.thingsboard.rule.engine.api.TbContext; @@ -27,6 +29,7 @@ import org.thingsboard.rule.engine.api.TbNode; import org.thingsboard.rule.engine.api.TbNodeConfiguration; import org.thingsboard.rule.engine.api.TbNodeException; import org.thingsboard.rule.engine.api.util.TbNodeUtils; +import org.thingsboard.rule.engine.util.SemaphoreWithTbMsgQueue; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.kv.TsKvEntry; import org.thingsboard.server.common.data.msg.TbMsgType; @@ -37,15 +40,14 @@ import org.thingsboard.server.dao.timeseries.TimeseriesService; import java.math.BigDecimal; import java.math.RoundingMode; -import java.util.List; import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; import static org.thingsboard.common.util.DonAsynchron.withCallback; @Slf4j @RuleNode(type = ComponentType.ENRICHMENT, - name = "calculate delta", relationTypes = {TbNodeConnectionType.SUCCESS, TbNodeConnectionType.FAILURE, TbNodeConnectionType.OTHER}, + name = "calculate delta", + relationTypes = {TbNodeConnectionType.SUCCESS, TbNodeConnectionType.FAILURE, TbNodeConnectionType.OTHER}, configClazz = CalculateDeltaNodeConfiguration.class, nodeDescription = "Calculates delta and amount of time passed between previous timeseries key reading " + "and current value for this key from the incoming message", @@ -56,19 +58,24 @@ import static org.thingsboard.common.util.DonAsynchron.withCallback; public class CalculateDeltaNode implements TbNode { private Map cache; + private Map locks; + private CalculateDeltaNodeConfiguration config; private TbContext ctx; private TimeseriesService timeseriesService; private boolean useCache; + private String inputKey; @Override public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException { this.config = TbNodeUtils.convert(configuration, CalculateDeltaNodeConfiguration.class); this.ctx = ctx; this.timeseriesService = ctx.getTimeseriesService(); + this.inputKey = config.getInputValueKey(); this.useCache = config.isUseCache(); if (useCache) { - cache = new ConcurrentHashMap<>(); + locks = new ConcurrentReferenceHashMap<>(16, ConcurrentReferenceHashMap.ReferenceType.WEAK); + cache = new ConcurrentReferenceHashMap<>(16, ConcurrentReferenceHashMap.ReferenceType.SOFT); } } @@ -79,78 +86,34 @@ public class CalculateDeltaNode implements TbNode { return; } JsonNode json = JacksonUtil.toJsonNode(msg.getData()); - String inputKey = config.getInputValueKey(); if (!json.has(inputKey)) { ctx.tellNext(msg, TbNodeConnectionType.OTHER); return; } - withCallback(getLastValue(msg.getOriginator()), + if (useCache) { + var semaphoreWithQueue = locks.computeIfAbsent(msg.getOriginator(), SemaphoreWithTbMsgQueue::new); + semaphoreWithQueue.addToQueueAndTryProcess(msg, ctx, this::processMsgAsync); + return; + } + withCallback(fetchLatestValueAsync(msg.getOriginator()), previousData -> { - double currentValue = json.get(inputKey).asDouble(); - long currentTs = msg.getMetaDataTs(); - - if (useCache) { - cache.put(msg.getOriginator(), new ValueWithTs(currentTs, currentValue)); - } - - BigDecimal delta = BigDecimal.valueOf(previousData != null ? currentValue - previousData.value : 0.0); - - if (config.isTellFailureIfDeltaIsNegative() && delta.doubleValue() < 0) { - ctx.tellFailure(msg, new IllegalArgumentException("Delta value is negative!")); - return; - } - - if (config.getRound() != null) { - delta = delta.setScale(config.getRound(), RoundingMode.HALF_UP); - } - - ObjectNode result = (ObjectNode) json; - if (delta.stripTrailingZeros().scale() > 0) { - result.put(config.getOutputValueKey(), delta.doubleValue()); - } else { - result.put(config.getOutputValueKey(), delta.longValueExact()); - } - - if (config.isAddPeriodBetweenMsgs()) { - long period = previousData != null ? currentTs - previousData.ts : 0; - result.put(config.getPeriodValueKey(), period); - } - ctx.tellSuccess(TbMsg.transformMsgData(msg, JacksonUtil.toString(result))); + processCalculateDelta(msg.getOriginator(), msg.getMetaDataTs(), (ObjectNode) json, previousData); + ctx.tellSuccess(TbMsg.transformMsgData(msg, JacksonUtil.toString(json))); }, - t -> ctx.tellFailure(msg, t), ctx.getDbCallbackExecutor()); + t -> ctx.tellFailure(msg, t), MoreExecutors.directExecutor()); } @Override public void destroy() { if (useCache) { cache.clear(); + locks.clear(); } } private ListenableFuture fetchLatestValueAsync(EntityId entityId) { return Futures.transform(timeseriesService.findLatest(ctx.getTenantId(), entityId, config.getInputValueKey()), - tsKvEntryOpt -> tsKvEntryOpt.map(this::extractValue).orElse(null) - , ctx.getDbCallbackExecutor()); - } - - private ValueWithTs fetchLatestValue(EntityId entityId) { - List tsKvEntries = timeseriesService.findLatestSync( - ctx.getTenantId(), - entityId, - List.of(config.getInputValueKey())); - return extractValue(tsKvEntries.get(0)); - } - - private ListenableFuture getLastValue(EntityId entityId) { - if (useCache) { - ValueWithTs latestValue; - if ((latestValue = cache.get(entityId)) == null) { - latestValue = fetchLatestValue(entityId); - } - return Futures.immediateFuture(latestValue); - } else { - return fetchLatestValueAsync(entityId); - } + tsKvEntryOpt -> tsKvEntryOpt.map(this::extractValue).orElse(null), ctx.getDbCallbackExecutor()); } private ValueWithTs extractValue(TsKvEntry kvEntry) { @@ -176,6 +139,44 @@ public class CalculateDeltaNode implements TbNode { return new ValueWithTs(ts, result); } + private void processCalculateDelta(EntityId originator, long msgTs, ObjectNode json, ValueWithTs previousData) { + double currentValue = json.get(inputKey).asDouble(); + if (useCache) { + cache.put(originator, new ValueWithTs(msgTs, currentValue)); + } + BigDecimal delta = BigDecimal.valueOf(previousData != null ? currentValue - previousData.value : 0.0); + if (config.isTellFailureIfDeltaIsNegative() && delta.doubleValue() < 0) { + throw new IllegalArgumentException("Delta value is negative!"); + } + if (config.getRound() != null) { + delta = delta.setScale(config.getRound(), RoundingMode.HALF_UP); + } + if (delta.stripTrailingZeros().scale() > 0) { + json.put(config.getOutputValueKey(), delta.doubleValue()); + } else { + json.put(config.getOutputValueKey(), delta.longValueExact()); + } + if (config.isAddPeriodBetweenMsgs()) { + long period = previousData != null ? msgTs - previousData.ts : 0; + json.put(config.getPeriodValueKey(), period); + } + } + + protected ListenableFuture processMsgAsync(TbContext ctx, TbMsg msg) { + ListenableFuture latestValueFuture = getLatestFromCacheOrFetchFromDb(msg); + return Futures.transform(latestValueFuture, previousData -> { + ObjectNode json = (ObjectNode) JacksonUtil.toJsonNode(msg.getData()); + processCalculateDelta(msg.getOriginator(), msg.getMetaDataTs(), json, previousData); + return TbMsg.transformMsgData(msg, JacksonUtil.toString(json)); + }, MoreExecutors.directExecutor()); + } + + private ListenableFuture getLatestFromCacheOrFetchFromDb(TbMsg msg) { + EntityId originator = msg.getOriginator(); + ValueWithTs valueWithTs = cache.get(msg.getOriginator()); + return valueWithTs != null ? Futures.immediateFuture(valueWithTs) : fetchLatestValueAsync(originator); + } + private record ValueWithTs(long ts, double value) { } diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/util/SemaphoreWithTbMsgQueue.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/util/SemaphoreWithTbMsgQueue.java new file mode 100644 index 0000000000..39b502a59d --- /dev/null +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/util/SemaphoreWithTbMsgQueue.java @@ -0,0 +1,140 @@ +/** + * Copyright © 2016-2024 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.rule.engine.util; + +import com.google.common.util.concurrent.ListenableFuture; +import lombok.Data; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.thingsboard.common.util.DonAsynchron; +import org.thingsboard.rule.engine.api.TbContext; +import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.msg.TbMsg; + +import java.util.Queue; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.Semaphore; +import java.util.function.BiFunction; + +/** + * A utility class designed to manage a queue of messages for a specific entity, ensuring that + * message processing is synchronized on a per-entity basis. This is achieved through the use of a semaphore, + * allowing only one message at a time to be processed for each entity ID, thus preventing race conditions + * and ensuring thread-safe operations. + *

+ * This class is especially useful in scenarios where the order of message processing and + * resource access synchronization are crucial, such as updating caches or databases in a concurrent environment. + */ +@Data +@Slf4j +@RequiredArgsConstructor +public class SemaphoreWithTbMsgQueue { + + private final EntityId entityId; + private final Semaphore semaphore = new Semaphore(1); + private final Queue queue = new ConcurrentLinkedQueue<>(); + + /** + * Adds a message to the queue for asynchronous processing and attempts to process the queue if possible. + * This method is thread-safe and ensures that messages are processed in the order they were added, + * with each message for a specific entity being processed one at a time due to the semaphore control. + * + * @param msg The message to be processed. + * @param ctx The context in which the message should be processed. + * @param msgProcessingFunction The function that defines how the message will be processed. + */ + public void addToQueueAndTryProcess(TbMsg msg, TbContext ctx, BiFunction> msgProcessingFunction) { + queue.add(new TbMsgTbContextBiFunction(msg, ctx, msgProcessingFunction)); + tryProcessQueue(); + } + + /** + * Attempts to process the next message in the queue. If the semaphore is available (indicating + * that no other message for the same entity is currently being processed), this method will + * acquire the semaphore and start processing the message. If the semaphore is not available, + * this method will return immediately, ensuring that messages are processed sequentially + * for each entity. + *

+ * This method is automatically called after adding a message to the queue to ensure + * that the queue is processed promptly. + */ + private void tryProcessQueue() { + while (!queue.isEmpty()) { + // The semaphore have to be acquired before EACH poll and released before NEXT poll. + // Otherwise, some message will remain unprocessed in queue + if (!semaphore.tryAcquire()) { + return; + } + TbMsgTbContextBiFunction tbMsgTbContext = null; + try { + tbMsgTbContext = queue.poll(); + if (tbMsgTbContext == null) { + semaphore.release(); + continue; + } + final TbMsg msg = tbMsgTbContext.getMsg(); + if (!msg.getCallback().isMsgValid()) { + log.trace("[{}] Skipping non-valid message [{}]", entityId, msg); + semaphore.release(); + continue; + } + //DO PROCESSING + final TbContext ctx = tbMsgTbContext.getCtx(); + final ListenableFuture resultMsgFuture = tbMsgTbContext.getBiFunction().apply(ctx, msg); + DonAsynchron.withCallback(resultMsgFuture, resultMsg -> { + try { + ctx.tellSuccess(resultMsg); + } finally { + semaphore.release(); + tryProcessQueue(); + } + }, t -> { + try { + ctx.tellFailure(msg, t); + } finally { + semaphore.release(); + tryProcessQueue(); + } + }, ctx.getDbCallbackExecutor()); + } catch (Throwable t) { + semaphore.release(); + if (tbMsgTbContext == null) { // if no message polled, the loop become infinite, will throw exception + log.error("[{}] Failed to process TbMsgTbContext queue", entityId, t); + throw t; + } + TbMsg msg = tbMsgTbContext.getMsg(); + TbContext ctx = tbMsgTbContext.getCtx(); + log.warn("[{}] Failed to process message: {}", entityId, msg, t); + ctx.tellFailure(msg, t); // you are not allowed to throw here, because queue will remain unprocessed + continue; // We are probably the last who process the queue. We have to continue poll until get successful callback or queue is empty + } + break; //submitted async exact one task. next poll will try on callback + } + } + + /** + * A utility class to hold the tuple of a {@link TbMsg}, {@link TbContext}, and the message processing function. + * This facilitates passing these three elements as a single object within the queue. + */ + @Data + @RequiredArgsConstructor + private static class TbMsgTbContextBiFunction { + private final TbMsg msg; + private final TbContext ctx; + private final BiFunction> biFunction; + } + +} diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/math/TbMathNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/math/TbMathNodeTest.java index e036e6333e..80e669847c 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/math/TbMathNodeTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/math/TbMathNodeTest.java @@ -20,7 +20,6 @@ import com.google.common.util.concurrent.Futures; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.tuple.Triple; import org.assertj.core.api.SoftAssertions; -import org.junit.Assert; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -69,7 +68,6 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotNull; -import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyDouble; @@ -78,9 +76,7 @@ import static org.mockito.ArgumentMatchers.argThat; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.BDDMockito.willAnswer; import static org.mockito.BDDMockito.willReturn; -import static org.mockito.BDDMockito.willReturn; import static org.mockito.BDDMockito.willThrow; -import static org.mockito.Mockito.lenient; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.spy; @@ -543,7 +539,7 @@ public class TbMathNodeTest { ArgumentCaptor tCaptor = ArgumentCaptor.forClass(Throwable.class); Mockito.verify(ctx, Mockito.timeout(5000)).tellFailure(eq(msg), tCaptor.capture()); - Assert.assertNotNull(tCaptor.getValue().getMessage()); + assertNotNull(tCaptor.getValue().getMessage()); } @Test @@ -558,7 +554,7 @@ public class TbMathNodeTest { ArgumentCaptor tCaptor = ArgumentCaptor.forClass(Throwable.class); Mockito.verify(ctx, Mockito.timeout(5000)).tellFailure(eq(msg), tCaptor.capture()); - Assert.assertNotNull(tCaptor.getValue().getMessage()); + assertNotNull(tCaptor.getValue().getMessage()); } @Test @@ -574,10 +570,10 @@ public class TbMathNodeTest { List slowMsgList = IntStream.range(0, 5) .mapToObj(x -> TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, originatorSlow, TbMsgMetaData.EMPTY, JacksonUtil.newObjectNode().put("a", 2).put("b", 2).toString())) - .collect(Collectors.toList()); + .toList(); List fastMsgList = IntStream.range(0, 2) .mapToObj(x -> TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, originatorFast, TbMsgMetaData.EMPTY, JacksonUtil.newObjectNode().put("a", 2).put("b", 2).toString())) - .collect(Collectors.toList()); + .toList(); assertThat(slowMsgList.size()).as("slow msgs >= rule-dispatcher pool size").isGreaterThanOrEqualTo(RULE_DISPATCHER_POOL_SIZE); @@ -714,7 +710,7 @@ public class TbMathNodeTest { }).given(node).onMsg(any(), any()); return Triple.of(ctx, resultKey, node); }) - .collect(Collectors.toList()); + .toList(); ctxNodes.forEach(ctxNode -> ruleEngineDispatcherExecutor.executeAsync(() -> ctxNode.getRight() .onMsg(ctxNode.getLeft(), TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, originator, TbMsgMetaData.EMPTY, "{\"a\":2,\"b\":2}")))); ctxNodes.forEach(ctxNode -> verify(ctxNode.getRight(), timeout(5000)).onMsg(eq(ctxNode.getLeft()), any())); diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNodeTest.java index 188dd2e4ec..0dc228235d 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNodeTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNodeTest.java @@ -16,13 +16,12 @@ package org.thingsboard.rule.engine.metadata; import com.google.common.util.concurrent.Futures; -import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; import org.assertj.core.api.Assertions; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; import org.mockito.ArgumentCaptor; -import org.mockito.ArgumentMatcher; import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; import org.thingsboard.common.util.JacksonUtil; @@ -46,7 +45,6 @@ import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsgMetaData; import org.thingsboard.server.dao.timeseries.TimeseriesService; -import java.util.List; import java.util.Optional; import java.util.UUID; @@ -58,14 +56,13 @@ import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyList; import static org.mockito.ArgumentMatchers.anySet; import static org.mockito.ArgumentMatchers.anyString; -import static org.mockito.ArgumentMatchers.argThat; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.never; import static org.mockito.Mockito.reset; -import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; +@Slf4j @ExtendWith(MockitoExtension.class) public class CalculateDeltaNodeTest { @@ -110,7 +107,7 @@ public class CalculateDeltaNodeTest { node.onMsg(ctxMock, msg); // THEN - verify(ctxMock, times(1)).tellNext(eq(msg), eq(TbNodeConnectionType.OTHER)); + verify(ctxMock).tellNext(eq(msg), eq(TbNodeConnectionType.OTHER)); verify(ctxMock, never()).tellSuccess(any()); verify(ctxMock, never()).tellFailure(any(), any()); } @@ -124,7 +121,7 @@ public class CalculateDeltaNodeTest { node.onMsg(ctxMock, msg); // THEN - verify(ctxMock, times(1)).tellNext(eq(msg), eq(TbNodeConnectionType.OTHER)); + verify(ctxMock).tellNext(eq(msg), eq(TbNodeConnectionType.OTHER)); verify(ctxMock, never()).tellSuccess(any()); verify(ctxMock, never()).tellFailure(any(), any()); } @@ -139,7 +136,7 @@ public class CalculateDeltaNodeTest { node.onMsg(ctxMock, msg); // THEN - verify(ctxMock, times(1)).tellNext(eq(msg), eq(TbNodeConnectionType.OTHER)); + verify(ctxMock).tellNext(eq(msg), eq(TbNodeConnectionType.OTHER)); verify(ctxMock, never()).tellSuccess(any()); verify(ctxMock, never()).tellFailure(any(), any()); } @@ -165,7 +162,7 @@ public class CalculateDeltaNodeTest { // THEN var actualMsgCaptor = ArgumentCaptor.forClass(TbMsg.class); - verify(ctxMock, times(1)).tellSuccess(actualMsgCaptor.capture()); + verify(ctxMock).tellSuccess(actualMsgCaptor.capture()); verify(ctxMock, never()).tellNext(any(), anyString()); verify(ctxMock, never()).tellNext(any(), anySet()); verify(ctxMock, never()).tellFailure(any(), any()); @@ -195,7 +192,7 @@ public class CalculateDeltaNodeTest { // THEN var actualMsgCaptor = ArgumentCaptor.forClass(TbMsg.class); - verify(ctxMock, times(1)).tellSuccess(actualMsgCaptor.capture()); + verify(ctxMock).tellSuccess(actualMsgCaptor.capture()); verify(ctxMock, never()).tellNext(any(), anyString()); verify(ctxMock, never()).tellNext(any(), anySet()); verify(ctxMock, never()).tellFailure(any(), any()); @@ -225,7 +222,7 @@ public class CalculateDeltaNodeTest { // THEN var actualMsgCaptor = ArgumentCaptor.forClass(TbMsg.class); - verify(ctxMock, times(1)).tellSuccess(actualMsgCaptor.capture()); + verify(ctxMock).tellSuccess(actualMsgCaptor.capture()); verify(ctxMock, never()).tellNext(any(), anyString()); verify(ctxMock, never()).tellNext(any(), anySet()); verify(ctxMock, never()).tellFailure(any(), any()); @@ -246,7 +243,7 @@ public class CalculateDeltaNodeTest { nodeConfiguration = new TbNodeConfiguration(JacksonUtil.valueToTree(config)); node.init(ctxMock, nodeConfiguration); - mockFindLatest(new BasicTsKvEntry(1L, new DoubleDataEntry("temperature", 40.0))); + mockFindLatestAsync(new BasicTsKvEntry(1L, new DoubleDataEntry("temperature", 40.0))); var msgData = "{\"temperature\": 42,\"airPressure\":123}"; var firstMsgMetaData = new TbMsgMetaData(); @@ -259,7 +256,7 @@ public class CalculateDeltaNodeTest { // THEN var actualMsgCaptor = ArgumentCaptor.forClass(TbMsg.class); - verify(ctxMock, times(1)).tellSuccess(actualMsgCaptor.capture()); + verify(ctxMock).tellSuccess(actualMsgCaptor.capture()); verify(ctxMock, never()).tellNext(any(), anyString()); verify(ctxMock, never()).tellNext(any(), anySet()); verify(ctxMock, never()).tellFailure(any(), any()); @@ -273,6 +270,8 @@ public class CalculateDeltaNodeTest { reset(ctxMock); reset(timeseriesServiceMock); + when(ctxMock.getDbCallbackExecutor()).thenReturn(DB_EXECUTOR); + var secondMsgMetaData = new TbMsgMetaData(); secondMsgMetaData.putValue("ts", String.valueOf(6L)); var secondMsg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, DUMMY_DEVICE_ORIGINATOR, secondMsgMetaData, msgData); @@ -284,7 +283,7 @@ public class CalculateDeltaNodeTest { actualMsgCaptor = ArgumentCaptor.forClass(TbMsg.class); verify(timeseriesServiceMock, never()).findLatest(any(), any(), anyList()); - verify(ctxMock, times(1)).tellSuccess(actualMsgCaptor.capture()); + verify(ctxMock).tellSuccess(actualMsgCaptor.capture()); verify(ctxMock, never()).tellNext(any(), anyString()); verify(ctxMock, never()).tellNext(any(), anySet()); verify(ctxMock, never()).tellFailure(any(), any()); @@ -314,7 +313,7 @@ public class CalculateDeltaNodeTest { // THEN var actualMsgCaptor = ArgumentCaptor.forClass(TbMsg.class); - verify(ctxMock, times(1)).tellSuccess(actualMsgCaptor.capture()); + verify(ctxMock).tellSuccess(actualMsgCaptor.capture()); verify(ctxMock, never()).tellNext(any(), anyString()); verify(ctxMock, never()).tellNext(any(), anySet()); verify(ctxMock, never()).tellFailure(any(), any()); @@ -331,7 +330,7 @@ public class CalculateDeltaNodeTest { nodeConfiguration = new TbNodeConfiguration(JacksonUtil.valueToTree(config)); node.init(ctxMock, nodeConfiguration); - mockFindLatest(new BasicTsKvEntry(System.currentTimeMillis(), new LongDataEntry("pulseCounter", 200L))); + mockFindLatestAsync(new BasicTsKvEntry(System.currentTimeMillis(), new LongDataEntry("pulseCounter", 200L))); var msgData = "{\"pulseCounter\":\"123\"}"; var msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, DUMMY_DEVICE_ORIGINATOR, TbMsgMetaData.EMPTY, msgData); @@ -343,7 +342,7 @@ public class CalculateDeltaNodeTest { var actualMsgCaptor = ArgumentCaptor.forClass(TbMsg.class); var actualExceptionCaptor = ArgumentCaptor.forClass(Exception.class); - verify(ctxMock, times(1)).tellFailure(actualMsgCaptor.capture(), actualExceptionCaptor.capture()); + verify(ctxMock).tellFailure(actualMsgCaptor.capture(), actualExceptionCaptor.capture()); verify(ctxMock, never()).tellSuccess(any()); verify(ctxMock, never()).tellNext(any(), anyString()); verify(ctxMock, never()).tellNext(any(), anySet()); @@ -363,7 +362,7 @@ public class CalculateDeltaNodeTest { nodeConfiguration = new TbNodeConfiguration(JacksonUtil.valueToTree(config)); node.init(ctxMock, nodeConfiguration); - mockFindLatest(new BasicTsKvEntry(System.currentTimeMillis(), new LongDataEntry("pulseCounter", 200L))); + mockFindLatestAsync(new BasicTsKvEntry(System.currentTimeMillis(), new LongDataEntry("pulseCounter", 200L))); var msgData = "{\"pulseCounter\":\"123\"}"; var msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, DUMMY_DEVICE_ORIGINATOR, TbMsgMetaData.EMPTY, msgData); @@ -374,7 +373,7 @@ public class CalculateDeltaNodeTest { // THEN var actualMsgCaptor = ArgumentCaptor.forClass(TbMsg.class); - verify(ctxMock, times(1)).tellSuccess(actualMsgCaptor.capture()); + verify(ctxMock).tellSuccess(actualMsgCaptor.capture()); verify(ctxMock, never()).tellFailure(any(), any()); verify(ctxMock, never()).tellNext(any(), anyString()); verify(ctxMock, never()).tellNext(any(), anySet()); @@ -386,13 +385,23 @@ public class CalculateDeltaNodeTest { @Test public void givenInvalidStringValue_whenOnMsg_thenException() { // GIVEN - mockFindLatest(new BasicTsKvEntry(System.currentTimeMillis(), new StringDataEntry("pulseCounter", "high"))); + mockFindLatestAsync(new BasicTsKvEntry(System.currentTimeMillis(), new StringDataEntry("pulseCounter", "high"))); var msgData = "{\"pulseCounter\":\"123\"}"; var msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, DUMMY_DEVICE_ORIGINATOR, TbMsgMetaData.EMPTY, msgData); - // WHEN-THEN - Assertions.assertThatThrownBy(() -> node.onMsg(ctxMock, msg)) + // WHEN + node.onMsg(ctxMock, msg); + + // THEN + ArgumentCaptor throwableCaptor = ArgumentCaptor.forClass(Throwable.class); + + verify(ctxMock).tellFailure(eq(msg), throwableCaptor.capture()); + verify(ctxMock, never()).tellSuccess(any()); + verify(ctxMock, never()).tellNext(any(), anyString()); + verify(ctxMock, never()).tellNext(any(), anySet()); + + Assertions.assertThat(throwableCaptor.getValue()) .isInstanceOf(IllegalArgumentException.class) .hasMessage("Calculation failed. Unable to parse value [high] of telemetry [pulseCounter] to Double"); } @@ -400,13 +409,23 @@ public class CalculateDeltaNodeTest { @Test public void givenBooleanValue_whenOnMsg_thenException() { // GIVEN - mockFindLatest(new BasicTsKvEntry(System.currentTimeMillis(), new BooleanDataEntry("pulseCounter", false))); + mockFindLatestAsync(new BasicTsKvEntry(System.currentTimeMillis(), new BooleanDataEntry("pulseCounter", false))); var msgData = "{\"pulseCounter\":true}"; var msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, DUMMY_DEVICE_ORIGINATOR, TbMsgMetaData.EMPTY, msgData); - // WHEN-THEN - Assertions.assertThatThrownBy(() -> node.onMsg(ctxMock, msg)) + // WHEN + node.onMsg(ctxMock, msg); + + // THEN + ArgumentCaptor throwableCaptor = ArgumentCaptor.forClass(Throwable.class); + + verify(ctxMock).tellFailure(eq(msg), throwableCaptor.capture()); + verify(ctxMock, never()).tellSuccess(any()); + verify(ctxMock, never()).tellNext(any(), anyString()); + verify(ctxMock, never()).tellNext(any(), anySet()); + + Assertions.assertThat(throwableCaptor.getValue()) .isInstanceOf(IllegalArgumentException.class) .hasMessage("Calculation failed. Boolean values are not supported!"); } @@ -414,24 +433,27 @@ public class CalculateDeltaNodeTest { @Test public void givenJsonValue_whenOnMsg_thenException() { // GIVEN - mockFindLatest(new BasicTsKvEntry(System.currentTimeMillis(), new JsonDataEntry("pulseCounter", "{\"isActive\":false}"))); + mockFindLatestAsync(new BasicTsKvEntry(System.currentTimeMillis(), new JsonDataEntry("pulseCounter", "{\"isActive\":false}"))); var msgData = "{\"pulseCounter\":{\"isActive\":true}}"; var msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, DUMMY_DEVICE_ORIGINATOR, TbMsgMetaData.EMPTY, msgData); - // WHEN-THEN - Assertions.assertThatThrownBy(() -> node.onMsg(ctxMock, msg)) + // WHEN + node.onMsg(ctxMock, msg); + + // THEN + ArgumentCaptor throwableCaptor = ArgumentCaptor.forClass(Throwable.class); + + verify(ctxMock).tellFailure(eq(msg), throwableCaptor.capture()); + verify(ctxMock, never()).tellSuccess(any()); + verify(ctxMock, never()).tellNext(any(), anyString()); + verify(ctxMock, never()).tellNext(any(), anySet()); + + Assertions.assertThat(throwableCaptor.getValue()) .isInstanceOf(IllegalArgumentException.class) .hasMessage("Calculation failed. JSON values are not supported!"); } - private void mockFindLatest(TsKvEntry tsKvEntry) { - when(ctxMock.getTenantId()).thenReturn(TENANT_ID); - when(timeseriesServiceMock.findLatestSync( - eq(TENANT_ID), eq(DUMMY_DEVICE_ORIGINATOR), argThat(new ListMatcher<>(List.of(tsKvEntry.getKey()))) - )).thenReturn(List.of(tsKvEntry)); - } - private void mockFindLatestAsync(TsKvEntry tsKvEntry) { when(ctxMock.getDbCallbackExecutor()).thenReturn(DB_EXECUTOR); when(ctxMock.getTenantId()).thenReturn(TENANT_ID); @@ -440,22 +462,4 @@ public class CalculateDeltaNodeTest { )).thenReturn(Futures.immediateFuture(Optional.of(tsKvEntry))); } - @RequiredArgsConstructor - private static class ListMatcher implements ArgumentMatcher> { - - private final List expectedList; - - @Override - public boolean matches(List actualList) { - if (actualList == expectedList) { - return true; - } - if (actualList.size() != expectedList.size()) { - return false; - } - return actualList.containsAll(expectedList); - } - - } - } From 827c898179e39f24bd4b6c283fb0a1ac1a2f083c Mon Sep 17 00:00:00 2001 From: ShvaykaD Date: Tue, 2 Apr 2024 12:38:37 +0300 Subject: [PATCH 3/5] added test givenConcurrentAccess_whenOnMsg_thenGetFromDBInvokedOnce --- .../metadata/CalculateDeltaNodeTest.java | 79 ++++++++++++++++++- 1 file changed, 75 insertions(+), 4 deletions(-) diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNodeTest.java index 0dc228235d..36b9719d86 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNodeTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNodeTest.java @@ -17,13 +17,13 @@ package org.thingsboard.rule.engine.metadata; import com.google.common.util.concurrent.Futures; import lombok.extern.slf4j.Slf4j; -import org.assertj.core.api.Assertions; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; import org.mockito.ArgumentCaptor; import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; +import org.thingsboard.common.util.AbstractListeningExecutor; import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.common.util.ListeningExecutor; import org.thingsboard.rule.engine.TestDbCallbackExecutor; @@ -45,9 +45,15 @@ import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsgMetaData; import org.thingsboard.server.dao.timeseries.TimeseriesService; +import java.util.List; import java.util.Optional; import java.util.UUID; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.stream.IntStream; +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertInstanceOf; @@ -57,8 +63,12 @@ import static org.mockito.ArgumentMatchers.anyList; import static org.mockito.ArgumentMatchers.anySet; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.BDDMockito.willAnswer; +import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.reset; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -69,6 +79,10 @@ public class CalculateDeltaNodeTest { private final DeviceId DUMMY_DEVICE_ORIGINATOR = new DeviceId(UUID.fromString("2ba3ded4-882b-40cf-999a-89da9ccd58f9")); private final TenantId TENANT_ID = new TenantId(UUID.fromString("3842e740-0d89-43a9-8d52-ae44023847ba")); private final ListeningExecutor DB_EXECUTOR = new TestDbCallbackExecutor(); + + private static final int RULE_DISPATCHER_POOL_SIZE = 2; + private static final int DB_CALLBACK_POOL_SIZE = 3; + @Mock private TbContext ctxMock; @Mock @@ -401,7 +415,7 @@ public class CalculateDeltaNodeTest { verify(ctxMock, never()).tellNext(any(), anyString()); verify(ctxMock, never()).tellNext(any(), anySet()); - Assertions.assertThat(throwableCaptor.getValue()) + assertThat(throwableCaptor.getValue()) .isInstanceOf(IllegalArgumentException.class) .hasMessage("Calculation failed. Unable to parse value [high] of telemetry [pulseCounter] to Double"); } @@ -425,7 +439,7 @@ public class CalculateDeltaNodeTest { verify(ctxMock, never()).tellNext(any(), anyString()); verify(ctxMock, never()).tellNext(any(), anySet()); - Assertions.assertThat(throwableCaptor.getValue()) + assertThat(throwableCaptor.getValue()) .isInstanceOf(IllegalArgumentException.class) .hasMessage("Calculation failed. Boolean values are not supported!"); } @@ -449,11 +463,68 @@ public class CalculateDeltaNodeTest { verify(ctxMock, never()).tellNext(any(), anyString()); verify(ctxMock, never()).tellNext(any(), anySet()); - Assertions.assertThat(throwableCaptor.getValue()) + assertThat(throwableCaptor.getValue()) .isInstanceOf(IllegalArgumentException.class) .hasMessage("Calculation failed. JSON values are not supported!"); } + @Test + public void givenConcurrentAccess_whenOnMsg_thenGetFromDBInvokedOnce() throws TbNodeException, InterruptedException { + DBCallbackExecutor dbCallbackExecutor = new DBCallbackExecutor(); + dbCallbackExecutor.init(); + + RuleDispatcherExecutor ruleEngineDispatcherExecutor = new RuleDispatcherExecutor(); + ruleEngineDispatcherExecutor.init(); + + assertThat(RULE_DISPATCHER_POOL_SIZE).as("dispatcher pool size have to be > 1").isGreaterThan(1); + + final TbContext ctx = mock(TbContext.class); + final TimeseriesService timeseriesService = mock(TimeseriesService.class); + + when(ctx.getTimeseriesService()).thenReturn(timeseriesService); + when(ctx.getDbCallbackExecutor()).thenReturn(dbCallbackExecutor); + when(timeseriesService.findLatest(any(), any(), anyString())).thenReturn(Futures.immediateFuture(Optional.empty())); + + final CalculateDeltaNodeConfiguration config = new CalculateDeltaNodeConfiguration().defaultConfiguration(); + final TbNodeConfiguration nodeConfiguration = new TbNodeConfiguration(JacksonUtil.valueToTree(config)); + final CalculateDeltaNode node = spy(CalculateDeltaNode.class); + + node.init(ctx, nodeConfiguration); + + List tbMsgList = IntStream.range(0, RULE_DISPATCHER_POOL_SIZE * 2).mapToObj(x -> { + var msgData = "{\"pulseCounter\":" + 2 + "}"; + return TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, DUMMY_DEVICE_ORIGINATOR, TbMsgMetaData.EMPTY, msgData); + }).toList(); + + CountDownLatch processingLatch = new CountDownLatch(tbMsgList.size()); + + willAnswer(invocation -> { + processingLatch.countDown(); + return invocation.callRealMethod(); + }).given(node).processMsgAsync(any(), any()); + + tbMsgList.forEach(msg -> ruleEngineDispatcherExecutor.executeAsync(() -> node.onMsg(ctx, msg))); + + assertThat(processingLatch.await(5, TimeUnit.SECONDS)).as("await on processingLatch").isTrue(); + + verify(timeseriesService).findLatest(any(), any(), anyString()); + await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> verify(ctx, times(tbMsgList.size())).tellSuccess(any())); + } + + private static class RuleDispatcherExecutor extends AbstractListeningExecutor { + @Override + protected int getThreadPollSize() { + return RULE_DISPATCHER_POOL_SIZE; + } + } + + private static class DBCallbackExecutor extends AbstractListeningExecutor { + @Override + protected int getThreadPollSize() { + return DB_CALLBACK_POOL_SIZE; + } + } + private void mockFindLatestAsync(TsKvEntry tsKvEntry) { when(ctxMock.getDbCallbackExecutor()).thenReturn(DB_EXECUTOR); when(ctxMock.getTenantId()).thenReturn(TENANT_ID); From 27d026821af4fb5da6206e12e689ef3e00edd1ce Mon Sep 17 00:00:00 2001 From: ShvaykaD Date: Tue, 16 Apr 2024 11:56:56 +0300 Subject: [PATCH 4/5] fixed adter review --- .../dao/timeseries/TimeseriesService.java | 2 - .../dao/sqlts/SqlTimeseriesLatestDao.java | 11 +- .../dao/timeseries/BaseTimeseriesService.java | 11 -- .../CassandraBaseTimeseriesLatestDao.java | 11 -- .../dao/timeseries/TimeseriesLatestDao.java | 2 - .../rule/engine/math/TbMathNode.java | 4 +- .../engine/metadata/CalculateDeltaNode.java | 102 ++++++++---------- .../metadata/CalculateDeltaNodeTest.java | 51 ++++++++- 8 files changed, 98 insertions(+), 96 deletions(-) diff --git a/common/dao-api/src/main/java/org/thingsboard/server/dao/timeseries/TimeseriesService.java b/common/dao-api/src/main/java/org/thingsboard/server/dao/timeseries/TimeseriesService.java index c62898ab09..ffea217da7 100644 --- a/common/dao-api/src/main/java/org/thingsboard/server/dao/timeseries/TimeseriesService.java +++ b/common/dao-api/src/main/java/org/thingsboard/server/dao/timeseries/TimeseriesService.java @@ -42,8 +42,6 @@ public interface TimeseriesService { ListenableFuture> findLatest(TenantId tenantId, EntityId entityId, Collection keys); - List findLatestSync(TenantId tenantId, EntityId entityId, Collection keys); - ListenableFuture> findAllLatest(TenantId tenantId, EntityId entityId); ListenableFuture save(TenantId tenantId, EntityId entityId, TsKvEntry tsKvEntry); diff --git a/dao/src/main/java/org/thingsboard/server/dao/sqlts/SqlTimeseriesLatestDao.java b/dao/src/main/java/org/thingsboard/server/dao/sqlts/SqlTimeseriesLatestDao.java index a9fb85560a..425bb10a0e 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sqlts/SqlTimeseriesLatestDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sqlts/SqlTimeseriesLatestDao.java @@ -19,6 +19,8 @@ import com.google.common.collect.Lists; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.MoreExecutors; +import jakarta.annotation.PostConstruct; +import jakarta.annotation.PreDestroy; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; @@ -50,8 +52,6 @@ import org.thingsboard.server.dao.sqlts.latest.TsKvLatestRepository; import org.thingsboard.server.dao.timeseries.TimeseriesLatestDao; import org.thingsboard.server.dao.util.SqlTsLatestAnyDao; -import jakarta.annotation.PostConstruct; -import jakarta.annotation.PreDestroy; import java.util.ArrayList; import java.util.Comparator; import java.util.HashMap; @@ -163,11 +163,6 @@ public class SqlTimeseriesLatestDao extends BaseAbstractSqlTimeseriesDao impleme return service.submit(() -> getLatestTsKvEntry(entityId, key)); } - @Override - public TsKvEntry findLatestSync(TenantId tenantId, EntityId entityId, String key) { - return getLatestTsKvEntry(entityId, key); - } - @Override public ListenableFuture> findAllLatest(TenantId tenantId, EntityId entityId) { return getFindAllLatestFuture(entityId); @@ -209,7 +204,7 @@ public class SqlTimeseriesLatestDao extends BaseAbstractSqlTimeseriesDao impleme ReadTsKvQueryResult::getData, MoreExecutors.directExecutor()); } - protected TsKvEntry doFindLatest(EntityId entityId, String key) { + protected TsKvEntry doFindLatest(EntityId entityId, String key) { TsKvLatestCompositeKey compositeKey = new TsKvLatestCompositeKey( entityId.getId(), diff --git a/dao/src/main/java/org/thingsboard/server/dao/timeseries/BaseTimeseriesService.java b/dao/src/main/java/org/thingsboard/server/dao/timeseries/BaseTimeseriesService.java index cfba6786f3..60056b2b8f 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/timeseries/BaseTimeseriesService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/timeseries/BaseTimeseriesService.java @@ -134,17 +134,6 @@ public class BaseTimeseriesService implements TimeseriesService { return Futures.allAsList(futures); } - @Override - public List findLatestSync(TenantId tenantId, EntityId entityId, Collection keys) { - validate(entityId); - List latestEntries = new ArrayList<>(keys.size()); - keys.forEach(key -> Validator.validateString(key, k -> "Incorrect key " + k)); - for (String key : keys) { - latestEntries.add(timeseriesLatestDao.findLatestSync(tenantId, entityId, key)); - } - return latestEntries; - } - @Override public ListenableFuture> findAllLatest(TenantId tenantId, EntityId entityId) { validate(entityId); diff --git a/dao/src/main/java/org/thingsboard/server/dao/timeseries/CassandraBaseTimeseriesLatestDao.java b/dao/src/main/java/org/thingsboard/server/dao/timeseries/CassandraBaseTimeseriesLatestDao.java index 342d4d9771..7a5904eb6b 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/timeseries/CassandraBaseTimeseriesLatestDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/timeseries/CassandraBaseTimeseriesLatestDao.java @@ -44,7 +44,6 @@ import org.thingsboard.server.dao.util.NoSqlTsLatestDao; import java.util.Collections; import java.util.List; import java.util.Optional; -import java.util.concurrent.ExecutionException; import static com.datastax.oss.driver.api.querybuilder.QueryBuilder.literal; @@ -70,16 +69,6 @@ public class CassandraBaseTimeseriesLatestDao extends AbstractCassandraBaseTimes return findLatest(tenantId, entityId, key, rs -> convertResultToTsKvEntry(key, rs.one())); } - @Override - public TsKvEntry findLatestSync(TenantId tenantId, EntityId entityId, String key) { - try { - return findLatest(tenantId, entityId, key, rs -> convertResultToTsKvEntry(key, rs.one())).get(); - } catch (InterruptedException | ExecutionException e) { - log.error("[{}][{}] Failed to get latest entry for key: {} due to: ", tenantId, entityId, key, e); - throw new RuntimeException(e); - } - } - private ListenableFuture findLatest(TenantId tenantId, EntityId entityId, String key, java.util.function.Function function) { BoundStatementBuilder stmtBuilder = new BoundStatementBuilder(getFindLatestStmt().bind()); stmtBuilder.setString(0, entityId.getEntityType().name()); diff --git a/dao/src/main/java/org/thingsboard/server/dao/timeseries/TimeseriesLatestDao.java b/dao/src/main/java/org/thingsboard/server/dao/timeseries/TimeseriesLatestDao.java index aad4074f85..d339f49e11 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/timeseries/TimeseriesLatestDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/timeseries/TimeseriesLatestDao.java @@ -40,8 +40,6 @@ public interface TimeseriesLatestDao { */ ListenableFuture findLatest(TenantId tenantId, EntityId entityId, String key); - TsKvEntry findLatestSync(TenantId tenantId, EntityId entityId, String key); - ListenableFuture> findAllLatest(TenantId tenantId, EntityId entityId); ListenableFuture saveLatest(TenantId tenantId, EntityId entityId, TsKvEntry tsKvEntry); diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/math/TbMathNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/math/TbMathNode.java index 823c224833..4e64ad854e 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/math/TbMathNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/math/TbMathNode.java @@ -105,8 +105,8 @@ public class TbMathNode implements TbNode { @Override public void onMsg(TbContext ctx, TbMsg msg) { - var semaphoreWithQueue = locks.computeIfAbsent(msg.getOriginator(), SemaphoreWithTbMsgQueue::new); - semaphoreWithQueue.addToQueueAndTryProcess(msg, ctx, this::processMsgAsync); + locks.computeIfAbsent(msg.getOriginator(), SemaphoreWithTbMsgQueue::new) + .addToQueueAndTryProcess(msg, ctx, this::processMsgAsync); } ListenableFuture processMsgAsync(TbContext ctx, TbMsg msg) { diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNode.java index f459a532ed..1552558a9e 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNode.java @@ -30,20 +30,18 @@ import org.thingsboard.rule.engine.api.TbNodeConfiguration; import org.thingsboard.rule.engine.api.TbNodeException; import org.thingsboard.rule.engine.api.util.TbNodeUtils; import org.thingsboard.rule.engine.util.SemaphoreWithTbMsgQueue; +import org.thingsboard.server.common.data.StringUtils; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.kv.TsKvEntry; import org.thingsboard.server.common.data.msg.TbMsgType; import org.thingsboard.server.common.data.msg.TbNodeConnectionType; import org.thingsboard.server.common.data.plugin.ComponentType; import org.thingsboard.server.common.msg.TbMsg; -import org.thingsboard.server.dao.timeseries.TimeseriesService; import java.math.BigDecimal; import java.math.RoundingMode; import java.util.Map; -import static org.thingsboard.common.util.DonAsynchron.withCallback; - @Slf4j @RuleNode(type = ComponentType.ENRICHMENT, name = "calculate delta", @@ -61,20 +59,21 @@ public class CalculateDeltaNode implements TbNode { private Map locks; private CalculateDeltaNodeConfiguration config; - private TbContext ctx; - private TimeseriesService timeseriesService; - private boolean useCache; - private String inputKey; @Override public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException { this.config = TbNodeUtils.convert(configuration, CalculateDeltaNodeConfiguration.class); - this.ctx = ctx; - this.timeseriesService = ctx.getTimeseriesService(); - this.inputKey = config.getInputValueKey(); - this.useCache = config.isUseCache(); - if (useCache) { - locks = new ConcurrentReferenceHashMap<>(16, ConcurrentReferenceHashMap.ReferenceType.WEAK); + if (StringUtils.isBlank(config.getInputValueKey())) { + throw new TbNodeException("Input value key should be specified!", true); + } + if (StringUtils.isBlank(config.getOutputValueKey())) { + throw new TbNodeException("Output value key should be specified!", true); + } + if (config.isAddPeriodBetweenMsgs() && StringUtils.isBlank(config.getPeriodValueKey())) { + throw new TbNodeException("Period value key should be specified!", true); + } + locks = new ConcurrentReferenceHashMap<>(16, ConcurrentReferenceHashMap.ReferenceType.WEAK); + if (config.isUseCache()) { cache = new ConcurrentReferenceHashMap<>(16, ConcurrentReferenceHashMap.ReferenceType.SOFT); } } @@ -85,35 +84,26 @@ public class CalculateDeltaNode implements TbNode { ctx.tellNext(msg, TbNodeConnectionType.OTHER); return; } - JsonNode json = JacksonUtil.toJsonNode(msg.getData()); - if (!json.has(inputKey)) { + JsonNode msgData = JacksonUtil.toJsonNode(msg.getData()); + if (msgData == null || !msgData.has(config.getInputValueKey())) { ctx.tellNext(msg, TbNodeConnectionType.OTHER); return; } - if (useCache) { - var semaphoreWithQueue = locks.computeIfAbsent(msg.getOriginator(), SemaphoreWithTbMsgQueue::new); - semaphoreWithQueue.addToQueueAndTryProcess(msg, ctx, this::processMsgAsync); - return; - } - withCallback(fetchLatestValueAsync(msg.getOriginator()), - previousData -> { - processCalculateDelta(msg.getOriginator(), msg.getMetaDataTs(), (ObjectNode) json, previousData); - ctx.tellSuccess(TbMsg.transformMsgData(msg, JacksonUtil.toString(json))); - }, - t -> ctx.tellFailure(msg, t), MoreExecutors.directExecutor()); + locks.computeIfAbsent(msg.getOriginator(), SemaphoreWithTbMsgQueue::new) + .addToQueueAndTryProcess(msg, ctx, this::processMsgAsync); } @Override public void destroy() { - if (useCache) { + locks.clear(); + if (config.isUseCache()) { cache.clear(); - locks.clear(); } } - private ListenableFuture fetchLatestValueAsync(EntityId entityId) { - return Futures.transform(timeseriesService.findLatest(ctx.getTenantId(), entityId, config.getInputValueKey()), - tsKvEntryOpt -> tsKvEntryOpt.map(this::extractValue).orElse(null), ctx.getDbCallbackExecutor()); + private ListenableFuture fetchLatestValueAsync(TbContext ctx, EntityId entityId) { + return Futures.transform(ctx.getTimeseriesService().findLatest(ctx.getTenantId(), entityId, config.getInputValueKey()), + tsKvEntryOpt -> tsKvEntryOpt.map(this::extractValue).orElse(null), MoreExecutors.directExecutor()); } private ValueWithTs extractValue(TsKvEntry kvEntry) { @@ -139,42 +129,38 @@ public class CalculateDeltaNode implements TbNode { return new ValueWithTs(ts, result); } - private void processCalculateDelta(EntityId originator, long msgTs, ObjectNode json, ValueWithTs previousData) { - double currentValue = json.get(inputKey).asDouble(); - if (useCache) { - cache.put(originator, new ValueWithTs(msgTs, currentValue)); - } - BigDecimal delta = BigDecimal.valueOf(previousData != null ? currentValue - previousData.value : 0.0); - if (config.isTellFailureIfDeltaIsNegative() && delta.doubleValue() < 0) { - throw new IllegalArgumentException("Delta value is negative!"); - } - if (config.getRound() != null) { - delta = delta.setScale(config.getRound(), RoundingMode.HALF_UP); - } - if (delta.stripTrailingZeros().scale() > 0) { - json.put(config.getOutputValueKey(), delta.doubleValue()); - } else { - json.put(config.getOutputValueKey(), delta.longValueExact()); - } - if (config.isAddPeriodBetweenMsgs()) { - long period = previousData != null ? msgTs - previousData.ts : 0; - json.put(config.getPeriodValueKey(), period); - } - } - protected ListenableFuture processMsgAsync(TbContext ctx, TbMsg msg) { - ListenableFuture latestValueFuture = getLatestFromCacheOrFetchFromDb(msg); + ListenableFuture latestValueFuture = getLatestFromCacheOrFetchFromDb(ctx, msg); return Futures.transform(latestValueFuture, previousData -> { ObjectNode json = (ObjectNode) JacksonUtil.toJsonNode(msg.getData()); - processCalculateDelta(msg.getOriginator(), msg.getMetaDataTs(), json, previousData); + double currentValue = json.get(config.getInputValueKey()).asDouble(); + if (config.isUseCache()) { + cache.put(msg.getOriginator(), new ValueWithTs(msg.getMetaDataTs(), currentValue)); + } + BigDecimal delta = BigDecimal.valueOf(previousData != null ? currentValue - previousData.value : 0.0); + if (config.isTellFailureIfDeltaIsNegative() && delta.doubleValue() < 0) { + throw new IllegalArgumentException("Delta value is negative!"); + } + if (config.getRound() != null) { + delta = delta.setScale(config.getRound(), RoundingMode.HALF_UP); + } + if (delta.stripTrailingZeros().scale() > 0) { + json.put(config.getOutputValueKey(), delta.doubleValue()); + } else { + json.put(config.getOutputValueKey(), delta.longValueExact()); + } + if (config.isAddPeriodBetweenMsgs()) { + long period = previousData != null ? msg.getMetaDataTs() - previousData.ts : 0; + json.put(config.getPeriodValueKey(), period); + } return TbMsg.transformMsgData(msg, JacksonUtil.toString(json)); }, MoreExecutors.directExecutor()); } - private ListenableFuture getLatestFromCacheOrFetchFromDb(TbMsg msg) { + private ListenableFuture getLatestFromCacheOrFetchFromDb(TbContext ctx, TbMsg msg) { EntityId originator = msg.getOriginator(); ValueWithTs valueWithTs = cache.get(msg.getOriginator()); - return valueWithTs != null ? Futures.immediateFuture(valueWithTs) : fetchLatestValueAsync(originator); + return valueWithTs != null ? Futures.immediateFuture(valueWithTs) : fetchLatestValueAsync(ctx, originator); } private record ValueWithTs(long ts, double value) { diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNodeTest.java index 36b9719d86..93e3957c03 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNodeTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNodeTest.java @@ -20,6 +20,9 @@ import lombok.extern.slf4j.Slf4j; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.NullAndEmptySource; +import org.junit.jupiter.params.provider.ValueSource; import org.mockito.ArgumentCaptor; import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; @@ -54,9 +57,11 @@ import java.util.stream.IntStream; import static org.assertj.core.api.Assertions.assertThat; import static org.awaitility.Awaitility.await; +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyList; @@ -96,8 +101,6 @@ public class CalculateDeltaNodeTest { node = new CalculateDeltaNode(); config = new CalculateDeltaNodeConfiguration().defaultConfiguration(); nodeConfiguration = new TbNodeConfiguration(JacksonUtil.valueToTree(config)); - when(ctxMock.getTimeseriesService()).thenReturn(timeseriesServiceMock); - node.init(ctxMock, nodeConfiguration); } @@ -111,6 +114,49 @@ public class CalculateDeltaNodeTest { assertTrue(config.isTellFailureIfDeltaIsNegative()); } + + @ParameterizedTest + @NullAndEmptySource + @ValueSource(strings = {" "}) // blank value + public void givenInvalidInputKey_whenInitThenThrowException(String key) { + config.setInputValueKey(key); + nodeConfiguration = new TbNodeConfiguration(JacksonUtil.valueToTree(config)); + var exception = assertThrows(TbNodeException.class, () -> node.init(ctxMock, nodeConfiguration)); + assertThat(exception).hasMessage("Input value key should be specified!"); + assertThat(exception.isUnrecoverable()).isTrue(); + } + + @ParameterizedTest + @NullAndEmptySource + @ValueSource(strings = {" "}) // blank value + public void givenInvalidOutputKey_whenInitThenThrowException(String key) { + config.setOutputValueKey(key); + nodeConfiguration = new TbNodeConfiguration(JacksonUtil.valueToTree(config)); + var exception = assertThrows(TbNodeException.class, () -> node.init(ctxMock, nodeConfiguration)); + assertThat(exception).hasMessage("Output value key should be specified!"); + assertThat(exception.isUnrecoverable()).isTrue(); + } + + @ParameterizedTest + @NullAndEmptySource + @ValueSource(strings = {" "}) // blank value + public void givenInvalidPeriodKey_whenInitThenThrowException(String key) { + config.setPeriodValueKey(key); + config.setAddPeriodBetweenMsgs(true); + nodeConfiguration = new TbNodeConfiguration(JacksonUtil.valueToTree(config)); + var exception = assertThrows(TbNodeException.class, () -> node.init(ctxMock, nodeConfiguration)); + assertThat(exception).hasMessage("Period value key should be specified!"); + assertThat(exception.isUnrecoverable()).isTrue(); + } + + @Test + public void givenInvalidPeriodKeyAndAddPeriodDisabled_whenInitThenNoExceptionThrown() { + config.setPeriodValueKey(null); + config.setAddPeriodBetweenMsgs(false); + nodeConfiguration = new TbNodeConfiguration(JacksonUtil.valueToTree(config)); + assertDoesNotThrow(() -> node.init(ctxMock, nodeConfiguration)); + } + @Test public void givenInvalidMsgType_whenOnMsg_thenShouldTellNextOther() { // GIVEN @@ -528,6 +574,7 @@ public class CalculateDeltaNodeTest { private void mockFindLatestAsync(TsKvEntry tsKvEntry) { when(ctxMock.getDbCallbackExecutor()).thenReturn(DB_EXECUTOR); when(ctxMock.getTenantId()).thenReturn(TENANT_ID); + when(ctxMock.getTimeseriesService()).thenReturn(timeseriesServiceMock); when(timeseriesServiceMock.findLatest( eq(TENANT_ID), eq(DUMMY_DEVICE_ORIGINATOR), eq(tsKvEntry.getKey()) )).thenReturn(Futures.immediateFuture(Optional.of(tsKvEntry))); From daa3c766f0f6f8f1bb11fd6b83d96befaa91d689 Mon Sep 17 00:00:00 2001 From: ShvaykaD Date: Tue, 16 Apr 2024 12:08:28 +0300 Subject: [PATCH 5/5] additional fixes in utils classes and tests --- .../engine/util/SemaphoreWithTbMsgQueue.java | 22 +++++++------------ .../metadata/CalculateDeltaNodeTest.java | 2 +- 2 files changed, 9 insertions(+), 15 deletions(-) diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/util/SemaphoreWithTbMsgQueue.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/util/SemaphoreWithTbMsgQueue.java index 39b502a59d..fa00856b4b 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/util/SemaphoreWithTbMsgQueue.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/util/SemaphoreWithTbMsgQueue.java @@ -17,7 +17,6 @@ package org.thingsboard.rule.engine.util; import com.google.common.util.concurrent.ListenableFuture; import lombok.Data; -import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.thingsboard.common.util.DonAsynchron; import org.thingsboard.rule.engine.api.TbContext; @@ -40,7 +39,6 @@ import java.util.function.BiFunction; */ @Data @Slf4j -@RequiredArgsConstructor public class SemaphoreWithTbMsgQueue { private final EntityId entityId; @@ -85,15 +83,15 @@ public class SemaphoreWithTbMsgQueue { semaphore.release(); continue; } - final TbMsg msg = tbMsgTbContext.getMsg(); + final TbMsg msg = tbMsgTbContext.msg(); if (!msg.getCallback().isMsgValid()) { log.trace("[{}] Skipping non-valid message [{}]", entityId, msg); semaphore.release(); continue; } //DO PROCESSING - final TbContext ctx = tbMsgTbContext.getCtx(); - final ListenableFuture resultMsgFuture = tbMsgTbContext.getBiFunction().apply(ctx, msg); + final TbContext ctx = tbMsgTbContext.ctx(); + final ListenableFuture resultMsgFuture = tbMsgTbContext.biFunction().apply(ctx, msg); DonAsynchron.withCallback(resultMsgFuture, resultMsg -> { try { ctx.tellSuccess(resultMsg); @@ -115,8 +113,8 @@ public class SemaphoreWithTbMsgQueue { log.error("[{}] Failed to process TbMsgTbContext queue", entityId, t); throw t; } - TbMsg msg = tbMsgTbContext.getMsg(); - TbContext ctx = tbMsgTbContext.getCtx(); + TbMsg msg = tbMsgTbContext.msg(); + TbContext ctx = tbMsgTbContext.ctx(); log.warn("[{}] Failed to process message: {}", entityId, msg, t); ctx.tellFailure(msg, t); // you are not allowed to throw here, because queue will remain unprocessed continue; // We are probably the last who process the queue. We have to continue poll until get successful callback or queue is empty @@ -126,15 +124,11 @@ public class SemaphoreWithTbMsgQueue { } /** - * A utility class to hold the tuple of a {@link TbMsg}, {@link TbContext}, and the message processing function. + * A utility record to hold the tuple of a {@link TbMsg}, {@link TbContext}, and the message processing function. * This facilitates passing these three elements as a single object within the queue. */ - @Data - @RequiredArgsConstructor - private static class TbMsgTbContextBiFunction { - private final TbMsg msg; - private final TbContext ctx; - private final BiFunction> biFunction; + private record TbMsgTbContextBiFunction(TbMsg msg, TbContext ctx, + BiFunction> biFunction) { } } diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNodeTest.java index 93e3957c03..0606030f30 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNodeTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNodeTest.java @@ -82,7 +82,7 @@ import static org.mockito.Mockito.when; public class CalculateDeltaNodeTest { private final DeviceId DUMMY_DEVICE_ORIGINATOR = new DeviceId(UUID.fromString("2ba3ded4-882b-40cf-999a-89da9ccd58f9")); - private final TenantId TENANT_ID = new TenantId(UUID.fromString("3842e740-0d89-43a9-8d52-ae44023847ba")); + private final TenantId TENANT_ID = TenantId.fromUUID(UUID.fromString("3842e740-0d89-43a9-8d52-ae44023847ba")); private final ListeningExecutor DB_EXECUTOR = new TestDbCallbackExecutor(); private static final int RULE_DISPATCHER_POOL_SIZE = 2;