diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java index cae377cc2d..fe02335223 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java +++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java @@ -248,7 +248,9 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor {ctx.tellNext(m, SUCCESS); sentTickMsg(ctx);}, - t -> {ctx.tellFailure(msg, t); sentTickMsg(ctx);}); + m -> { + ctx.tellNext(m, SUCCESS); + scheduleTickMsg(ctx); + }, + t -> { + ctx.tellFailure(msg, t); + scheduleTickMsg(ctx); + }); } } - private void sentTickMsg(TbContext ctx) { + private void scheduleTickMsg(TbContext ctx) { + long curTs = System.currentTimeMillis(); + if (lastScheduledTs == 0L) { + lastScheduledTs = curTs; + } + lastScheduledTs = lastScheduledTs + delay; + long curDelay = Math.max(0L, (lastScheduledTs - curTs)); TbMsg tickMsg = ctx.newMsg(TB_MSG_GENERATOR_NODE_MSG, ctx.getSelfId(), new TbMsgMetaData(), ""); nextTickId = tickMsg.getId(); - ctx.tellSelf(tickMsg, delay); + ctx.tellSelf(tickMsg, curDelay); } private ListenableFuture generate(TbContext ctx) { return ctx.getJsExecutor().executeAsync(() -> { if (prevMsg == null) { - prevMsg = ctx.newMsg( "", originatorId, new TbMsgMetaData(), "{}"); + prevMsg = ctx.newMsg("", originatorId, new TbMsgMetaData(), "{}"); } TbMsg generated = jsEngine.executeGenerate(prevMsg); prevMsg = ctx.newMsg(generated.getType(), originatorId, generated.getMetaData(), generated.getData());