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

additional fixes in utils classes and tests

Browse Source
This commit is contained in:
ShvaykaD 2024-04-16 12:08:28 +03:00
parent 27d026821a
commit daa3c766f0
2 changed files with 9 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -17,7 +17,6 @@ package org.thingsboard.rule.engine.util;
import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.ListenableFuture;
import lombok.Data; import lombok.Data;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.thingsboard.common.util.DonAsynchron; import org.thingsboard.common.util.DonAsynchron;
import org.thingsboard.rule.engine.api.TbContext; import org.thingsboard.rule.engine.api.TbContext;
@ -40,7 +39,6 @@ import java.util.function.BiFunction;
*/ */
@Data @Data
@Slf4j @Slf4j
@RequiredArgsConstructor
public class SemaphoreWithTbMsgQueue { public class SemaphoreWithTbMsgQueue {
private final EntityId entityId; private final EntityId entityId;
@ -85,15 +83,15 @@ public class SemaphoreWithTbMsgQueue {
semaphore.release(); semaphore.release();
continue; continue;
} }
final TbMsg msg = tbMsgTbContext.getMsg(); final TbMsg msg = tbMsgTbContext.msg();
if (!msg.getCallback().isMsgValid()) { if (!msg.getCallback().isMsgValid()) {
log.trace("[{}] Skipping non-valid message [{}]", entityId, msg); log.trace("[{}] Skipping non-valid message [{}]", entityId, msg);
semaphore.release(); semaphore.release();
continue; continue;
} }
//DO PROCESSING //DO PROCESSING
final TbContext ctx = tbMsgTbContext.getCtx(); final TbContext ctx = tbMsgTbContext.ctx();
final ListenableFuture<TbMsg> resultMsgFuture = tbMsgTbContext.getBiFunction().apply(ctx, msg); final ListenableFuture<TbMsg> resultMsgFuture = tbMsgTbContext.biFunction().apply(ctx, msg);
DonAsynchron.withCallback(resultMsgFuture, resultMsg -> { DonAsynchron.withCallback(resultMsgFuture, resultMsg -> {
try { try {
ctx.tellSuccess(resultMsg); ctx.tellSuccess(resultMsg);
@ -115,8 +113,8 @@ public class SemaphoreWithTbMsgQueue {
log.error("[{}] Failed to process TbMsgTbContext queue", entityId, t); log.error("[{}] Failed to process TbMsgTbContext queue", entityId, t);
throw t; throw t;
} }
TbMsg msg = tbMsgTbContext.getMsg(); TbMsg msg = tbMsgTbContext.msg();
TbContext ctx = tbMsgTbContext.getCtx(); TbContext ctx = tbMsgTbContext.ctx();
log.warn("[{}] Failed to process message: {}", entityId, msg, t); 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 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 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. * This facilitates passing these three elements as a single object within the queue.
*/ */
@Data private record TbMsgTbContextBiFunction(TbMsg msg, TbContext ctx,
@RequiredArgsConstructor BiFunction<TbContext, TbMsg, ListenableFuture<TbMsg>> biFunction) {
private static class TbMsgTbContextBiFunction {
private final TbMsg msg;
private final TbContext ctx;
private final BiFunction<TbContext, TbMsg, ListenableFuture<TbMsg>> biFunction;
} }
} }

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

@ -82,7 +82,7 @@ import static org.mockito.Mockito.when;
public class CalculateDeltaNodeTest { public class CalculateDeltaNodeTest {
private final DeviceId DUMMY_DEVICE_ORIGINATOR = new DeviceId(UUID.fromString("2ba3ded4-882b-40cf-999a-89da9ccd58f9")); 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 final ListeningExecutor DB_EXECUTOR = new TestDbCallbackExecutor();
private static final int RULE_DISPATCHER_POOL_SIZE = 2; private static final int RULE_DISPATCHER_POOL_SIZE = 2;