dendi-wang/thingsboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

TbMathNode: call processMsgAsync via BiFunction this::processMsgAsync

Browse Source
This commit is contained in:
Sergey Matvienko 2023-08-31 00:07:35 +02:00
parent 168d0980f0
commit c28e4157e7
2 changed files with 8 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -82,7 +82,7 @@ import java.util.stream.Collectors;
)
public class TbMathNode implements TbNode {
private static final ConcurrentMap<EntityId, SemaphoreWithQueue<TbMsgTbContext>> locks = new ConcurrentReferenceHashMap<>(16, ConcurrentReferenceHashMap.ReferenceType.WEAK);
private static final ConcurrentMap<EntityId, SemaphoreWithQueue<TbMsgTbContextBiFunction>> locks = new ConcurrentReferenceHashMap<>(16, ConcurrentReferenceHashMap.ReferenceType.WEAK);
private final ThreadLocal<Expression> customExpression = new ThreadLocal<>();
private TbMathNodeConfiguration config;
private boolean msgBodyToJsonConversionRequired;
@ -109,21 +109,21 @@ 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 TbMsgTbContext(msg, ctx));
semaphoreWithQueue.getQueue().add(new TbMsgTbContextBiFunction(msg, ctx, this::processMsgAsync));
tryProcessQueue(semaphoreWithQueue);
}
void tryProcessQueue(SemaphoreWithQueue<TbMsgTbContext> lockAndQueue) {
void tryProcessQueue(SemaphoreWithQueue<TbMsgTbContextBiFunction> lockAndQueue) {
final Semaphore semaphore = lockAndQueue.getSemaphore();
final Queue<TbMsgTbContext> queue = lockAndQueue.getQueue();
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;
}
TbMsgTbContext tbMsgTbContext = null;
TbMsgTbContextBiFunction tbMsgTbContext = null;
try {
tbMsgTbContext = queue.poll();
if (tbMsgTbContext == null) {
@ -138,7 +138,7 @@ public class TbMathNode implements TbNode {
}
//DO PROCESSING
final TbContext ctx = tbMsgTbContext.getCtx();
final ListenableFuture<TbMsg> resultMsgFuture = processMsgAsync(ctx, msg);
final ListenableFuture<TbMsg> resultMsgFuture = tbMsgTbContext.getBiFunction().apply(ctx, msg);
DonAsynchron.withCallback(resultMsgFuture, resultMsg -> {
try {
ctx.tellSuccess(resultMsg);
@ -432,9 +432,10 @@ public class TbMathNode implements TbNode {
@Data
@RequiredArgsConstructor
static public class TbMsgTbContext {
static public class TbMsgTbContextBiFunction {
final TbMsg msg;
final TbContext ctx;
final BiFunction<TbContext, TbMsg, ListenableFuture<TbMsg>> biFunction;
}
}

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

@ -652,7 +652,6 @@ public class TbMathNodeTest {
List<Triple<TbContext, String, TbMathNode>> ctxNodes = IntStream.range(0, RULE_DISPATCHER_POOL_SIZE * 2)
.mapToObj(x -> {
final TbContext ctx = mock(TbContext.class); // many rule nodes - many contexts
willReturn(tenantId).given(ctx).getTenantId();
willReturn(dbCallbackExecutor).given(ctx).getDbCallbackExecutor();
final String resultKey = "result" + x;
final TbMathNode node = spy(initNodeWithCustomFunction(ctx, "2a+3b",