tsai
/
thingsboard
mirror of https://github.com/thingsboard/thingsboard.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

Merge pull request #10483 from ShvaykaD/improvements/calculate-delta-node 

CalculateDeltaNode make non-blocking on find latest telemetry
pull/10592/head
Viacheslav Klimov 2 years ago
committed by GitHub
parent
6c7ed5ee90 c92294610b
commit
cbda6154f8
No known key found for this signature in database GPG Key ID: B5690EEEBB952194
10 changed files with 405 additions and 305 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 4
      common/dao-api/src/main/java/org/thingsboard/server/dao/timeseries/TimeseriesService.java
  2. 11
      dao/src/main/java/org/thingsboard/server/dao/sqlts/SqlTimeseriesLatestDao.java
  3. 11
      dao/src/main/java/org/thingsboard/server/dao/timeseries/BaseTimeseriesService.java
  4. 11
      dao/src/main/java/org/thingsboard/server/dao/timeseries/CassandraBaseTimeseriesLatestDao.java
  5. 2
      dao/src/main/java/org/thingsboard/server/dao/timeseries/TimeseriesLatestDao.java
  6. 91
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/math/TbMathNode.java
  7. 170
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNode.java
  8. 134
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/util/SemaphoreWithTbMsgQueue.java
  9. 14
      rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/math/TbMathNodeTest.java
  10. 262
      rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNodeTest.java

4
common/dao-api/src/main/java/org/thingsboard/server/dao/timeseries/TimeseriesService.java
View File

@ -38,12 +38,10 @@ public interface TimeseriesService {
ListenableFuture<List<TsKvEntry>> findAll(TenantId tenantId, EntityId entityId, List<ReadTsKvQuery> queries);
ListenableFuture<Optional<TsKvEntry>> findLatest(TenantId tenantId, EntityId entityId, String keys);
ListenableFuture<Optional<TsKvEntry>> findLatest(TenantId tenantId, EntityId entityId, String key);
ListenableFuture<List<TsKvEntry>> findLatest(TenantId tenantId, EntityId entityId, Collection<String> keys);
List<TsKvEntry> findLatestSync(TenantId tenantId, EntityId entityId, Collection<String> keys);
ListenableFuture<List<TsKvEntry>> findAllLatest(TenantId tenantId, EntityId entityId);
ListenableFuture<Integer> save(TenantId tenantId, EntityId entityId, TsKvEntry tsKvEntry);

11
dao/src/main/java/org/thingsboard/server/dao/sqlts/SqlTimeseriesLatestDao.java
View File

@ -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<List<TsKvEntry>> 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(),

11
dao/src/main/java/org/thingsboard/server/dao/timeseries/BaseTimeseriesService.java
View File

@ -134,17 +134,6 @@ public class BaseTimeseriesService implements TimeseriesService {
return Futures.allAsList(futures);
}
@Override
public List<TsKvEntry> findLatestSync(TenantId tenantId, EntityId entityId, Collection<String> keys) {
validate(entityId);
List<TsKvEntry> 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<List<TsKvEntry>> findAllLatest(TenantId tenantId, EntityId entityId) {
validate(entityId);

11
dao/src/main/java/org/thingsboard/server/dao/timeseries/CassandraBaseTimeseriesLatestDao.java
View File

@ -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 <T> ListenableFuture<T> findLatest(TenantId tenantId, EntityId entityId, String key, java.util.function.Function<TbResultSet, T> function) {
BoundStatementBuilder stmtBuilder = new BoundStatementBuilder(getFindLatestStmt().bind());
stmtBuilder.setString(0, entityId.getEntityType().name());

2
dao/src/main/java/org/thingsboard/server/dao/timeseries/TimeseriesLatestDao.java
View File

@ -40,8 +40,6 @@ public interface TimeseriesLatestDao {
*/
ListenableFuture<TsKvEntry> findLatest(TenantId tenantId, EntityId entityId, String key);
TsKvEntry findLatestSync(TenantId tenantId, EntityId entityId, String key);
ListenableFuture<List<TsKvEntry>> findAllLatest(TenantId tenantId, EntityId entityId);
ListenableFuture<Void> saveLatest(TenantId tenantId, EntityId entityId, TsKvEntry tsKvEntry);

91
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/math/TbMathNode.java
View File

@ -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<EntityId, SemaphoreWithQueue<TbMsgTbContextBiFunction>> locks = new ConcurrentReferenceHashMap<>(16, ConcurrentReferenceHashMap.ReferenceType.WEAK);
private static final ConcurrentMap<EntityId, SemaphoreWithTbMsgQueue> locks = new ConcurrentReferenceHashMap<>(16, ConcurrentReferenceHashMap.ReferenceType.WEAK);
private final ThreadLocal<Expression> 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<TbMsgTbContextBiFunction> lockAndQueue) {
final Semaphore semaphore = lockAndQueue.getSemaphore();
final Queue<TbMsgTbContextBiFunction> 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<TbMsg> 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
}
locks.computeIfAbsent(msg.getOriginator(), SemaphoreWithTbMsgQueue::new)
.addToQueueAndTryProcess(msg, ctx, this::processMsgAsync);
}
ListenableFuture<TbMsg> processMsgAsync(TbContext ctx, TbMsg msg) {
@ -426,24 +363,4 @@ public class TbMathNode implements TbNode {
}
}
@Override
public void destroy() {
}
@Data
@RequiredArgsConstructor
static public class SemaphoreWithQueue<T> {
final EntityId entityId;
final Semaphore semaphore = new Semaphore(1);
final Queue<T> queue = new ConcurrentLinkedQueue<>();
}
@Data
@RequiredArgsConstructor
static public class TbMsgTbContextBiFunction {
final TbMsg msg;
final TbContext ctx;
final BiFunction<TbContext, TbMsg, ListenableFuture<TbMsg>> biFunction;
}
}

170
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNode.java
View File

@ -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,8 @@ 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.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;
@ -34,16 +38,10 @@ import org.thingsboard.server.common.data.msg.TbNodeConnectionType;
import org.thingsboard.server.common.data.plugin.ComponentType;
import org.thingsboard.server.common.data.util.TbPair;
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.Collections;
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,
@ -60,19 +58,25 @@ import static org.thingsboard.common.util.DonAsynchron.withCallback;
public class CalculateDeltaNode implements TbNode {
private Map<EntityId, ValueWithTs> cache;
private Map<EntityId, SemaphoreWithTbMsgQueue> locks;
private CalculateDeltaNodeConfiguration config;
private TbContext ctx;
private TimeseriesService timeseriesService;
private boolean useCache;
@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.useCache = config.isUseCache();
if (useCache) {
cache = new ConcurrentHashMap<>();
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);
}
}
@ -82,56 +86,19 @@ public class CalculateDeltaNode implements TbNode {
ctx.tellNext(msg, TbNodeConnectionType.OTHER);
return;
}
JsonNode json = JacksonUtil.toJsonNode(msg.getData());
String inputKey = config.getInputValueKey();
if (!json.has(inputKey)) {
JsonNode msgData = JacksonUtil.toJsonNode(msg.getData());
if (msgData == null || !msgData.has(config.getInputValueKey())) {
ctx.tellNext(msg, TbNodeConnectionType.OTHER);
return;
}
withCallback(getLastValue(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.isExcludeZeroDeltas() && delta.doubleValue() == 0) {
ctx.tellSuccess(msg);
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)));
},
t -> ctx.tellFailure(msg, t), ctx.getDbCallbackExecutor());
locks.computeIfAbsent(msg.getOriginator(), SemaphoreWithTbMsgQueue::new)
.addToQueueAndTryProcess(msg, ctx, this::processMsgAsync);
}
@Override
public void destroy() {
if (useCache) {
locks.clear();
if (config.isUseCache()) {
cache.clear();
}
}
@ -153,30 +120,9 @@ public class CalculateDeltaNode implements TbNode {
return new TbPair<>(hasChanges, oldConfiguration);
}
private ListenableFuture<ValueWithTs> fetchLatestValueAsync(EntityId entityId) {
return Futures.transform(timeseriesService.findLatest(ctx.getTenantId(), entityId, Collections.singletonList(config.getInputValueKey())),
list -> extractValue(list.get(0))
, ctx.getDbCallbackExecutor());
}
private ValueWithTs fetchLatestValue(EntityId entityId) {
List<TsKvEntry> tsKvEntries = timeseriesService.findLatestSync(
ctx.getTenantId(),
entityId,
Collections.singletonList(config.getInputValueKey()));
return extractValue(tsKvEntries.get(0));
}
private ListenableFuture<ValueWithTs> getLastValue(EntityId entityId) {
if (useCache) {
ValueWithTs latestValue;
if ((latestValue = cache.get(entityId)) == null) {
latestValue = fetchLatestValue(entityId);
}
return Futures.immediateFuture(latestValue);
} else {
return fetchLatestValueAsync(entityId);
}
private ListenableFuture<ValueWithTs> 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) {
@ -186,36 +132,60 @@ 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;
protected ListenableFuture<TbMsg> processMsgAsync(TbContext ctx, TbMsg msg) {
ListenableFuture<ValueWithTs> latestValueFuture = getLatestFromCacheOrFetchFromDb(ctx, msg);
return Futures.transform(latestValueFuture, previousData -> {
ObjectNode json = (ObjectNode) JacksonUtil.toJsonNode(msg.getData());
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.isExcludeZeroDeltas() && delta.doubleValue() == 0) {
return msg;
}
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<ValueWithTs> getLatestFromCacheOrFetchFromDb(TbContext ctx, TbMsg msg) {
EntityId originator = msg.getOriginator();
ValueWithTs valueWithTs = cache.get(msg.getOriginator());
return valueWithTs != null ? Futures.immediateFuture(valueWithTs) : fetchLatestValueAsync(ctx, originator);
}
private ValueWithTs(long ts, double value) {
this.ts = ts;
this.value = value;
}
private record ValueWithTs(long ts, double value) {
}
}

134
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/util/SemaphoreWithTbMsgQueue.java
View File

@ -0,0 +1,134 @@
/**
* 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.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.
* <p>
* 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
public class SemaphoreWithTbMsgQueue {
private final EntityId entityId;
private final Semaphore semaphore = new Semaphore(1);
private final Queue<TbMsgTbContextBiFunction> 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<TbContext, TbMsg, ListenableFuture<TbMsg>> 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.
* <p>
* 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.msg();
if (!msg.getCallback().isMsgValid()) {
log.trace("[{}] Skipping non-valid message [{}]", entityId, msg);
semaphore.release();
continue;
}
//DO PROCESSING
final TbContext ctx = tbMsgTbContext.ctx();
final ListenableFuture<TbMsg> resultMsgFuture = tbMsgTbContext.biFunction().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.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
}
break; //submitted async exact one task. next poll will try on callback
}
}
/**
* 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.
*/
private record TbMsgTbContextBiFunction(TbMsg msg, TbContext ctx,
BiFunction<TbContext, TbMsg, ListenableFuture<TbMsg>> biFunction) {
}
}

14
rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/math/TbMathNodeTest.java
View File

@ -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<Throwable> 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<Throwable> 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<TbMsg> 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<TbMsg> 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()));

262
rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNodeTest.java
View File

@ -16,20 +16,20 @@
package org.thingsboard.rule.engine.metadata;
import com.google.common.util.concurrent.Futures;
import lombok.Data;
import lombok.RequiredArgsConstructor;
import org.assertj.core.api.Assertions;
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.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
import org.junit.jupiter.params.provider.NullAndEmptySource;
import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.ArgumentCaptor;
import org.mockito.ArgumentMatcher;
import org.mockito.Mock;
import org.mockito.Spy;
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.AbstractRuleNodeUpgradeTest;
@ -54,34 +54,48 @@ 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.function.BiConsumer;
import java.util.stream.IntStream;
import java.util.stream.Stream;
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;
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.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.verifyNoMoreInteractions;
import static org.mockito.Mockito.when;
@Slf4j
@ExtendWith(MockitoExtension.class)
public class CalculateDeltaNodeTest extends AbstractRuleNodeUpgradeTest {
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 = 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;
private static final int DB_CALLBACK_POOL_SIZE = 3;
@Mock
private TbContext ctxMock;
@Mock
@ -95,8 +109,6 @@ public class CalculateDeltaNodeTest extends AbstractRuleNodeUpgradeTest {
public void setUp() throws TbNodeException {
config = new CalculateDeltaNodeConfiguration().defaultConfiguration();
nodeConfiguration = new TbNodeConfiguration(JacksonUtil.valueToTree(config));
when(ctxMock.getTimeseriesService()).thenReturn(timeseriesServiceMock);
node.init(ctxMock, nodeConfiguration);
}
@ -110,6 +122,49 @@ public class CalculateDeltaNodeTest extends AbstractRuleNodeUpgradeTest {
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
@ -120,7 +175,7 @@ public class CalculateDeltaNodeTest extends AbstractRuleNodeUpgradeTest {
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());
}
@ -134,7 +189,7 @@ public class CalculateDeltaNodeTest extends AbstractRuleNodeUpgradeTest {
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());
}
@ -149,7 +204,7 @@ public class CalculateDeltaNodeTest extends AbstractRuleNodeUpgradeTest {
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());
}
@ -175,7 +230,7 @@ public class CalculateDeltaNodeTest extends AbstractRuleNodeUpgradeTest {
// 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());
@ -205,7 +260,7 @@ public class CalculateDeltaNodeTest extends AbstractRuleNodeUpgradeTest {
// 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());
@ -235,7 +290,7 @@ public class CalculateDeltaNodeTest extends AbstractRuleNodeUpgradeTest {
// 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());
@ -256,7 +311,7 @@ public class CalculateDeltaNodeTest extends AbstractRuleNodeUpgradeTest {
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();
@ -269,7 +324,7 @@ public class CalculateDeltaNodeTest extends AbstractRuleNodeUpgradeTest {
// 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());
@ -283,6 +338,8 @@ public class CalculateDeltaNodeTest extends AbstractRuleNodeUpgradeTest {
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);
@ -294,7 +351,7 @@ public class CalculateDeltaNodeTest extends AbstractRuleNodeUpgradeTest {
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());
@ -324,7 +381,7 @@ public class CalculateDeltaNodeTest extends AbstractRuleNodeUpgradeTest {
// 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());
@ -341,7 +398,7 @@ public class CalculateDeltaNodeTest extends AbstractRuleNodeUpgradeTest {
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);
@ -353,7 +410,7 @@ public class CalculateDeltaNodeTest extends AbstractRuleNodeUpgradeTest {
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());
@ -373,7 +430,7 @@ public class CalculateDeltaNodeTest extends AbstractRuleNodeUpgradeTest {
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);
@ -384,7 +441,7 @@ public class CalculateDeltaNodeTest extends AbstractRuleNodeUpgradeTest {
// 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());
@ -396,13 +453,23 @@ public class CalculateDeltaNodeTest extends AbstractRuleNodeUpgradeTest {
@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<Throwable> 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());
assertThat(throwableCaptor.getValue())
.isInstanceOf(IllegalArgumentException.class)
.hasMessage("Calculation failed. Unable to parse value [high] of telemetry [pulseCounter] to Double");
}
@ -410,13 +477,23 @@ public class CalculateDeltaNodeTest extends AbstractRuleNodeUpgradeTest {
@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<Throwable> 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());
assertThat(throwableCaptor.getValue())
.isInstanceOf(IllegalArgumentException.class)
.hasMessage("Calculation failed. Boolean values are not supported!");
}
@ -424,38 +501,104 @@ public class CalculateDeltaNodeTest extends AbstractRuleNodeUpgradeTest {
@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<Throwable> 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());
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<TbMsg> 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;
}
}
@ParameterizedTest
@MethodSource("CalculateDeltaTestConfig")
public void givenCalculateDeltaConfig_whenOnMsg_thenVerify(CalculateDeltaTestConfig testConfig) throws TbNodeException {
// GIVEN
config.setTellFailureIfDeltaIsNegative(testConfig.isTellFailureIfDeltaIsNegative());
config.setExcludeZeroDeltas(testConfig.isExcludeZeroDeltas());
config.setTellFailureIfDeltaIsNegative(testConfig.tellFailureIfDeltaIsNegative());
config.setExcludeZeroDeltas(testConfig.excludeZeroDeltas());
config.setInputValueKey("temperature");
nodeConfiguration = new TbNodeConfiguration(JacksonUtil.valueToTree(config));
node.init(ctxMock, nodeConfiguration);
mockFindLatest(new BasicTsKvEntry(1L, new DoubleDataEntry("temperature", testConfig.getPrevValue())));
mockFindLatestAsync(new BasicTsKvEntry(1L, new DoubleDataEntry("temperature", testConfig.prevValue())));
var msgData = "{\"temperature\":" + testConfig.getCurrentValue() + ",\"airPressure\":123}";
var msgData = "{\"temperature\":" + testConfig.currentValue() + ",\"airPressure\":123}";
var msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, DUMMY_DEVICE_ORIGINATOR, TbMsgMetaData.EMPTY, msgData);
// WHEN
node.onMsg(ctxMock, msg);
// THEN
testConfig.getVerificationMethod().accept(ctxMock, msg);
testConfig.verificationMethod().accept(ctxMock, msg);
}
private static Stream<CalculateDeltaTestConfig> CalculateDeltaTestConfig() {
@ -510,47 +653,18 @@ public class CalculateDeltaNodeTest extends AbstractRuleNodeUpgradeTest {
);
}
@Data
@RequiredArgsConstructor
private static class CalculateDeltaTestConfig {
private final boolean tellFailureIfDeltaIsNegative;
private final boolean excludeZeroDeltas;
private final double prevValue;
private final double currentValue;
private final BiConsumer<TbContext, TbMsg> verificationMethod;
}
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 record CalculateDeltaTestConfig(boolean tellFailureIfDeltaIsNegative, boolean excludeZeroDeltas,
double prevValue, double currentValue,
BiConsumer<TbContext, TbMsg> verificationMethod) {
}
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), argThat(new ListMatcher<>(List.of(tsKvEntry.getKey())))
)).thenReturn(Futures.immediateFuture(List.of(tsKvEntry)));
}
@RequiredArgsConstructor
private static class ListMatcher<T> implements ArgumentMatcher<List<T>> {
private final List<T> expectedList;
@Override
public boolean matches(List<T> actualList) {
if (actualList == expectedList) {
return true;
}
if (actualList.size() != expectedList.size()) {
return false;
}
return actualList.containsAll(expectedList);
}
eq(TENANT_ID), eq(DUMMY_DEVICE_ORIGINATOR), eq(tsKvEntry.getKey())
)).thenReturn(Futures.immediateFuture(Optional.of(tsKvEntry)));
}
private static Stream<Arguments> givenFromVersionAndConfig_whenUpgrade_thenVerifyHasChangesAndConfig() {

Write Preview
Loading…
Cancel
Save