diff --git a/application/src/main/java/org/thingsboard/server/actors/device/DeviceActor.java b/application/src/main/java/org/thingsboard/server/actors/device/DeviceActor.java index 9c0215fc3a..feecef859f 100644 --- a/application/src/main/java/org/thingsboard/server/actors/device/DeviceActor.java +++ b/application/src/main/java/org/thingsboard/server/actors/device/DeviceActor.java @@ -20,6 +20,7 @@ import org.thingsboard.rule.engine.api.msg.DeviceAttributesEventNotificationMsg; import org.thingsboard.rule.engine.api.msg.DeviceNameOrTypeUpdateMsg; import org.thingsboard.server.actors.ActorSystemContext; import org.thingsboard.server.actors.TbActorCtx; +import org.thingsboard.server.actors.TbActorException; import org.thingsboard.server.actors.service.ContextAwareActor; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.TenantId; @@ -39,7 +40,7 @@ public class DeviceActor extends ContextAwareActor { } @Override - public void init(TbActorCtx ctx) { + public void init(TbActorCtx ctx) throws TbActorException { super.init(ctx); log.debug("[{}][{}] Starting device actor.", processor.tenantId, processor.deviceId); try { @@ -47,6 +48,7 @@ public class DeviceActor extends ContextAwareActor { log.debug("[{}][{}] Device actor started.", processor.tenantId, processor.deviceId); } catch (Exception e) { log.warn("[{}][{}] Unknown failure", processor.tenantId, processor.deviceId, e); + throw new TbActorException("Failed to initialize device actor", e); } } diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java index 8f831ada16..7610f89719 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java +++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java @@ -221,8 +221,8 @@ class DefaultTbContext implements TbContext { } @Override - public TbMsg newMsg(String type, EntityId originator, TbMsgMetaData metaData, String data) { - return TbMsg.newMsg(type, originator, metaData, data, nodeCtx.getSelf().getRuleChainId(), nodeCtx.getSelf().getId()); + public TbMsg newMsg(String queueName, String type, EntityId originator, TbMsgMetaData metaData, String data) { + return TbMsg.newMsg(queueName, type, originator, metaData, data, nodeCtx.getSelf().getRuleChainId(), nodeCtx.getSelf().getId()); } @Override 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 afb50a18aa..6beae6fac6 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 @@ -244,7 +244,7 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor relations = nodeRoutes.get(originatorNodeId).stream() .filter(r -> contains(relationTypes, r.getType())) .collect(Collectors.toList()); diff --git a/application/src/main/java/org/thingsboard/server/actors/service/ComponentActor.java b/application/src/main/java/org/thingsboard/server/actors/service/ComponentActor.java index da560cef97..43a4c0309d 100644 --- a/application/src/main/java/org/thingsboard/server/actors/service/ComponentActor.java +++ b/application/src/main/java/org/thingsboard/server/actors/service/ComponentActor.java @@ -17,7 +17,9 @@ package org.thingsboard.server.actors.service; import lombok.extern.slf4j.Slf4j; import org.thingsboard.server.actors.ActorSystemContext; +import org.thingsboard.server.actors.TbActor; import org.thingsboard.server.actors.TbActorCtx; +import org.thingsboard.server.actors.TbActorException; import org.thingsboard.server.actors.shared.ComponentMsgProcessor; import org.thingsboard.server.actors.stats.StatsPersistMsg; import org.thingsboard.server.common.data.id.EntityId; @@ -48,13 +50,13 @@ public abstract class ComponentActor getErrorPersistFrequency()) { diff --git a/application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java b/application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java index b46b5a25c0..b069a5b198 100644 --- a/application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java +++ b/application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java @@ -19,6 +19,7 @@ import lombok.extern.slf4j.Slf4j; import org.thingsboard.server.actors.ActorSystemContext; import org.thingsboard.server.actors.TbActor; import org.thingsboard.server.actors.TbActorCtx; +import org.thingsboard.server.actors.TbActorException; import org.thingsboard.server.actors.TbActorId; import org.thingsboard.server.actors.TbActorNotRegisteredException; import org.thingsboard.server.actors.TbActorRef; @@ -62,7 +63,7 @@ public class TenantActor extends RuleChainManagerActor { boolean cantFindTenant = false; @Override - public void init(TbActorCtx ctx) { + public void init(TbActorCtx ctx) throws TbActorException { super.init(ctx); log.info("[{}] Starting tenant actor.", tenantId); try { @@ -93,6 +94,8 @@ public class TenantActor extends RuleChainManagerActor { } } catch (Exception e) { log.warn("[{}] Unknown failure", tenantId, e); +// TODO: throw this in 3.1? +// throw new TbActorException("Failed to init actor", e); } } diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java index 6db8e73c2b..ff88f012ee 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java @@ -160,7 +160,7 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService log.warn("[{}] Timeout to process message: {}", id, msg.getValue())); + ctx.getAckMap().forEach((id, msg) -> log.debug("[{}] Timeout to process message: {}", id, msg.getValue())); ctx.getFailedMap().forEach((id, msg) -> log.warn("[{}] Failed to process message: {}", id, msg.getValue())); } mainConsumer.commit(); diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java index b98f043a0d..99a3abf2cc 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java @@ -166,7 +166,7 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService< TbMsgCallback callback = new TbMsgPackCallback(id, tenantId, ctx); try { if (toRuleEngineMsg.getTbMsg() != null && !toRuleEngineMsg.getTbMsg().isEmpty()) { - forwardToRuleEngineActor(tenantId, toRuleEngineMsg, callback); + forwardToRuleEngineActor(configuration.getName(), tenantId, toRuleEngineMsg, callback); } else { callback.onSuccess(); } @@ -180,7 +180,7 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService< timeout = true; } - TbRuleEngineProcessingResult result = new TbRuleEngineProcessingResult(timeout, ctx); + TbRuleEngineProcessingResult result = new TbRuleEngineProcessingResult(configuration.getName(), timeout, ctx); TbRuleEngineProcessingDecision decision = ackStrategy.analyze(result); if (statsEnabled) { stats.log(result, decision.isCommit()); @@ -246,8 +246,8 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService< } } - private void forwardToRuleEngineActor(TenantId tenantId, ToRuleEngineMsg toRuleEngineMsg, TbMsgCallback callback) { - TbMsg tbMsg = TbMsg.fromBytes(toRuleEngineMsg.getTbMsg().toByteArray(), callback); + private void forwardToRuleEngineActor(String queueName, TenantId tenantId, ToRuleEngineMsg toRuleEngineMsg, TbMsgCallback callback) { + TbMsg tbMsg = TbMsg.fromBytes(queueName, toRuleEngineMsg.getTbMsg().toByteArray(), callback); QueueToRuleEngineMsg msg; ProtocolStringList relationTypesList = toRuleEngineMsg.getRelationTypesList(); Set relationTypes = null; diff --git a/application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineProcessingResult.java b/application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineProcessingResult.java index e818ffb9d7..bbe717ce83 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineProcessingResult.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineProcessingResult.java @@ -27,6 +27,8 @@ import java.util.concurrent.ConcurrentMap; public class TbRuleEngineProcessingResult { + @Getter + private final String queueName; @Getter private final boolean success; @Getter @@ -34,7 +36,8 @@ public class TbRuleEngineProcessingResult { @Getter private final TbMsgPackProcessingContext ctx; - public TbRuleEngineProcessingResult(boolean timeout, TbMsgPackProcessingContext ctx) { + public TbRuleEngineProcessingResult(String queueName, boolean timeout, TbMsgPackProcessingContext ctx) { + this.queueName = queueName; this.timeout = timeout; this.ctx = ctx; this.success = !timeout && ctx.getPendingMap().isEmpty() && ctx.getFailedMap().isEmpty(); diff --git a/application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineProcessingStrategyFactory.java b/application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineProcessingStrategyFactory.java index 80b0523a81..b6220f5f94 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineProcessingStrategyFactory.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineProcessingStrategyFactory.java @@ -100,7 +100,7 @@ public class TbRuleEngineProcessingStrategyFactory { } log.debug("[{}] Going to reprocess {} messages", queueName, toReprocess.size()); if (log.isTraceEnabled()) { - toReprocess.forEach((id, msg) -> log.trace("Going to reprocess [{}]: {}", id, TbMsg.fromBytes(msg.getValue().getTbMsg().toByteArray(), TbMsgCallback.EMPTY))); + toReprocess.forEach((id, msg) -> log.trace("Going to reprocess [{}]: {}", id, TbMsg.fromBytes(result.getQueueName(), msg.getValue().getTbMsg().toByteArray(), TbMsgCallback.EMPTY))); } if (pauseBetweenRetries > 0) { try { @@ -129,10 +129,10 @@ public class TbRuleEngineProcessingStrategyFactory { log.debug("[{}] Reprocessing skipped for {} failed and {} timeout messages", queueName, result.getFailedMap().size(), result.getPendingMap().size()); } if (log.isTraceEnabled()) { - result.getFailedMap().forEach((id, msg) -> log.trace("Failed messages [{}]: {}", id, TbMsg.fromBytes(msg.getValue().getTbMsg().toByteArray(), TbMsgCallback.EMPTY))); + result.getFailedMap().forEach((id, msg) -> log.trace("Failed messages [{}]: {}", id, TbMsg.fromBytes(result.getQueueName(), msg.getValue().getTbMsg().toByteArray(), TbMsgCallback.EMPTY))); } if (log.isTraceEnabled()) { - result.getPendingMap().forEach((id, msg) -> log.trace("Timeout messages [{}]: {}", id, TbMsg.fromBytes(msg.getValue().getTbMsg().toByteArray(), TbMsgCallback.EMPTY))); + result.getPendingMap().forEach((id, msg) -> log.trace("Timeout messages [{}]: {}", id, TbMsg.fromBytes(result.getQueueName(), msg.getValue().getTbMsg().toByteArray(), TbMsgCallback.EMPTY))); } return new TbRuleEngineProcessingDecision(true, null); } diff --git a/application/src/main/java/org/thingsboard/server/service/script/AbstractNashornJsInvokeService.java b/application/src/main/java/org/thingsboard/server/service/script/AbstractNashornJsInvokeService.java index 5ea8d13270..8e6666bbba 100644 --- a/application/src/main/java/org/thingsboard/server/service/script/AbstractNashornJsInvokeService.java +++ b/application/src/main/java/org/thingsboard/server/service/script/AbstractNashornJsInvokeService.java @@ -126,7 +126,7 @@ public abstract class AbstractNashornJsInvokeService extends AbstractJsInvokeSer scriptIdToNameMap.put(scriptId, functionName); return scriptId; } catch (Exception e) { - log.warn("Failed to compile JS script: {}", e.getMessage(), e); + log.debug("Failed to compile JS script: {}", e.getMessage(), e); throw new ExecutionException(e); } }); diff --git a/common/actor/src/main/java/org/thingsboard/server/actors/AbstractTbActor.java b/common/actor/src/main/java/org/thingsboard/server/actors/AbstractTbActor.java index 2179070dd2..9ea2b6c67d 100644 --- a/common/actor/src/main/java/org/thingsboard/server/actors/AbstractTbActor.java +++ b/common/actor/src/main/java/org/thingsboard/server/actors/AbstractTbActor.java @@ -23,7 +23,7 @@ public abstract class AbstractTbActor implements TbActor { protected TbActorCtx ctx; @Override - public void init(TbActorCtx ctx) { + public void init(TbActorCtx ctx) throws TbActorException { this.ctx = ctx; } diff --git a/common/actor/src/main/java/org/thingsboard/server/actors/TbActor.java b/common/actor/src/main/java/org/thingsboard/server/actors/TbActor.java index 5fe1206a53..97adc5da93 100644 --- a/common/actor/src/main/java/org/thingsboard/server/actors/TbActor.java +++ b/common/actor/src/main/java/org/thingsboard/server/actors/TbActor.java @@ -23,14 +23,14 @@ public interface TbActor { TbActorRef getActorRef(); - default void init(TbActorCtx ctx) { + default void init(TbActorCtx ctx) throws TbActorException { } - default void destroy() { + default void destroy() throws TbActorException { } default InitFailureStrategy onInitFailure(int attempt, Throwable t) { - return InitFailureStrategy.retryWithDelay(5000); + return InitFailureStrategy.retryWithDelay(5000 * attempt); } default ProcessFailureStrategy onProcessFailure(Throwable t) { diff --git a/common/actor/src/main/java/org/thingsboard/server/actors/TbActorException.java b/common/actor/src/main/java/org/thingsboard/server/actors/TbActorException.java new file mode 100644 index 0000000000..c62fb5ac22 --- /dev/null +++ b/common/actor/src/main/java/org/thingsboard/server/actors/TbActorException.java @@ -0,0 +1,23 @@ +/** + * Copyright © 2016-2020 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.server.actors; + +public class TbActorException extends Exception { + + public TbActorException(String message, Throwable cause) { + super(message, cause); + } +} diff --git a/common/actor/src/main/java/org/thingsboard/server/actors/TbActorMailbox.java b/common/actor/src/main/java/org/thingsboard/server/actors/TbActorMailbox.java index 279332adf3..a0dd7482de 100644 --- a/common/actor/src/main/java/org/thingsboard/server/actors/TbActorMailbox.java +++ b/common/actor/src/main/java/org/thingsboard/server/actors/TbActorMailbox.java @@ -72,10 +72,12 @@ public final class TbActorMailbox implements TbActorCtx { log.info("[{}] Failed to init actor, attempt {}, going to stop attempts.", selfId, attempt, t); system.stop(selfId); } else if (strategy.getRetryDelay() > 0) { - log.info("[{}] Failed to init actor, attempt {}, going to retry in attempts in {}ms", selfId, attempt, strategy.getRetryDelay(), t); + log.info("[{}] Failed to init actor, attempt {}, going to retry in attempts in {}ms", selfId, attempt, strategy.getRetryDelay()); + log.debug("[{}] Error", selfId, t); system.getScheduler().schedule(() -> dispatcher.getExecutor().execute(() -> tryInit(attemptIdx)), strategy.getRetryDelay(), TimeUnit.MILLISECONDS); } else { - log.info("[{}] Failed to init actor, attempt {}, going to retry immediately", selfId, attempt, t); + log.info("[{}] Failed to init actor, attempt {}, going to retry immediately", selfId, attempt); + log.debug("[{}] Error", selfId, t); dispatcher.getExecutor().execute(() -> tryInit(attemptIdx)); } } diff --git a/common/actor/src/main/java/org/thingsboard/server/actors/TbEntityActorId.java b/common/actor/src/main/java/org/thingsboard/server/actors/TbEntityActorId.java index be3a1fc6d8..3a3e501c56 100644 --- a/common/actor/src/main/java/org/thingsboard/server/actors/TbEntityActorId.java +++ b/common/actor/src/main/java/org/thingsboard/server/actors/TbEntityActorId.java @@ -31,7 +31,7 @@ public class TbEntityActorId implements TbActorId { @Override public String toString() { - return entityId.toString(); + return entityId.getEntityType() + "|" + entityId.getId(); } @Override diff --git a/common/actor/src/test/java/org/thingsboard/server/actors/ActorSystemTest.java b/common/actor/src/test/java/org/thingsboard/server/actors/ActorSystemTest.java index 9b8329c2c8..5e0a9af548 100644 --- a/common/actor/src/test/java/org/thingsboard/server/actors/ActorSystemTest.java +++ b/common/actor/src/test/java/org/thingsboard/server/actors/ActorSystemTest.java @@ -148,6 +148,25 @@ public class ActorSystemTest { Assert.assertEquals(2, testCtx.getInvocationCount().get()); } + @Test + public void testFailedInit() throws InterruptedException { + actorSystem.createDispatcher(ROOT_DISPATCHER, Executors.newWorkStealingPool(parallelism)); + ActorTestCtx testCtx1 = getActorTestCtx(1); + ActorTestCtx testCtx2 = getActorTestCtx(1); + + TbActorRef actorId1 = actorSystem.createRootActor(ROOT_DISPATCHER, new FailedToInitActor.FailedToInitActorCreator( + new TbEntityActorId(new DeviceId(UUID.randomUUID())), testCtx1, 1, 3000)); + TbActorRef actorId2 = actorSystem.createRootActor(ROOT_DISPATCHER, new FailedToInitActor.FailedToInitActorCreator( + new TbEntityActorId(new DeviceId(UUID.randomUUID())), testCtx2, 2, 1)); + + actorId1.tell(new IntTbActorMsg(42)); + actorId2.tell(new IntTbActorMsg(42)); + + Assert.assertFalse(testCtx1.getLatch().await(2, TimeUnit.SECONDS)); + Assert.assertTrue(testCtx2.getLatch().await(1, TimeUnit.SECONDS)); + Assert.assertTrue(testCtx1.getLatch().await(3, TimeUnit.SECONDS)); + } + public void testActorsAndMessages(int actorsCount, int msgNumber, int times) throws InterruptedException { Random random = new Random(); diff --git a/common/actor/src/test/java/org/thingsboard/server/actors/FailedToInitActor.java b/common/actor/src/test/java/org/thingsboard/server/actors/FailedToInitActor.java new file mode 100644 index 0000000000..509c3e1af8 --- /dev/null +++ b/common/actor/src/test/java/org/thingsboard/server/actors/FailedToInitActor.java @@ -0,0 +1,72 @@ +/** + * Copyright © 2016-2020 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.server.actors; + +import lombok.extern.slf4j.Slf4j; + +@Slf4j +public class FailedToInitActor extends TestRootActor { + + int retryAttempts; + int retryDelay; + int attempts = 0; + + public FailedToInitActor(TbActorId actorId, ActorTestCtx testCtx, int retryAttempts, int retryDelay) { + super(actorId, testCtx); + this.retryAttempts = retryAttempts; + this.retryDelay = retryDelay; + } + + @Override + public void init(TbActorCtx ctx) throws TbActorException { + if (attempts < retryAttempts) { + attempts++; + throw new TbActorException("Test attempt", new RuntimeException()); + } else { + super.init(ctx); + } + } + + @Override + public InitFailureStrategy onInitFailure(int attempt, Throwable t) { + return InitFailureStrategy.retryWithDelay(retryDelay); + } + + public static class FailedToInitActorCreator implements TbActorCreator { + + private final TbActorId actorId; + private final ActorTestCtx testCtx; + private final int retryAttempts; + private final int retryDelay; + + public FailedToInitActorCreator(TbActorId actorId, ActorTestCtx testCtx, int retryAttempts, int retryDelay) { + this.actorId = actorId; + this.testCtx = testCtx; + this.retryAttempts = retryAttempts; + this.retryDelay = retryDelay; + } + + @Override + public TbActorId createActorId() { + return actorId; + } + + @Override + public TbActor createActor() { + return new FailedToInitActor(actorId, testCtx, retryAttempts, retryDelay); + } + } +} diff --git a/common/actor/src/test/java/org/thingsboard/server/actors/SlowInitActor.java b/common/actor/src/test/java/org/thingsboard/server/actors/SlowInitActor.java index e9ff1fd665..1b59b164ff 100644 --- a/common/actor/src/test/java/org/thingsboard/server/actors/SlowInitActor.java +++ b/common/actor/src/test/java/org/thingsboard/server/actors/SlowInitActor.java @@ -25,7 +25,7 @@ public class SlowInitActor extends TestRootActor { } @Override - public void init(TbActorCtx ctx) { + public void init(TbActorCtx ctx) throws TbActorException { try { Thread.sleep(500); } catch (InterruptedException e) { diff --git a/common/actor/src/test/java/org/thingsboard/server/actors/TestRootActor.java b/common/actor/src/test/java/org/thingsboard/server/actors/TestRootActor.java index 881e536b39..d4d56a0de5 100644 --- a/common/actor/src/test/java/org/thingsboard/server/actors/TestRootActor.java +++ b/common/actor/src/test/java/org/thingsboard/server/actors/TestRootActor.java @@ -37,7 +37,7 @@ public class TestRootActor extends AbstractTbActor { } @Override - public void init(TbActorCtx ctx) { + public void init(TbActorCtx ctx) throws TbActorException { super.init(ctx); initialized = true; } diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java b/common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java index 845138d92c..1604875c7b 100644 --- a/common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java +++ b/common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java @@ -25,6 +25,7 @@ import org.thingsboard.server.common.data.id.EntityIdFactory; import org.thingsboard.server.common.data.id.RuleChainId; import org.thingsboard.server.common.data.id.RuleNodeId; import org.thingsboard.server.common.msg.gen.MsgProtos; +import org.thingsboard.server.common.msg.queue.ServiceQueue; import org.thingsboard.server.common.msg.queue.TbMsgCallback; import java.io.IOException; @@ -39,6 +40,7 @@ import java.util.UUID; @Slf4j public final class TbMsg implements Serializable { + private final String queueName; private final UUID id; private final long ts; private final String type; @@ -51,39 +53,44 @@ public final class TbMsg implements Serializable { //This field is not serialized because we use queues and there is no need to do it transient private final TbMsgCallback callback; + public static TbMsg newMsg(String queueName, String type, EntityId originator, TbMsgMetaData metaData, String data, RuleChainId ruleChainId, RuleNodeId ruleNodeId) { + return new TbMsg(queueName, UUID.randomUUID(), System.currentTimeMillis(), type, originator, metaData.copy(), TbMsgDataType.JSON, data, ruleChainId, ruleNodeId, TbMsgCallback.EMPTY); + } + public static TbMsg newMsg(String type, EntityId originator, TbMsgMetaData metaData, String data) { - return new TbMsg(UUID.randomUUID(), System.currentTimeMillis(), type, originator, metaData.copy(), TbMsgDataType.JSON, data, null, null, TbMsgCallback.EMPTY); + return new TbMsg(ServiceQueue.MAIN, UUID.randomUUID(), System.currentTimeMillis(), type, originator, metaData.copy(), TbMsgDataType.JSON, data, null, null, TbMsgCallback.EMPTY); } - public static TbMsg newMsg(String type, EntityId originator, TbMsgMetaData metaData, String data, RuleChainId ruleChainId, RuleNodeId ruleNodeId) { - return new TbMsg(UUID.randomUUID(), System.currentTimeMillis(), type, originator, metaData.copy(), TbMsgDataType.JSON, data, ruleChainId, ruleNodeId, TbMsgCallback.EMPTY); + public static TbMsg newMsg(String queueName, String type, EntityId originator, TbMsgMetaData metaData, String data) { + return new TbMsg(queueName, UUID.randomUUID(), System.currentTimeMillis(), type, originator, metaData.copy(), TbMsgDataType.JSON, data, null, null, TbMsgCallback.EMPTY); } public static TbMsg newMsg(String type, EntityId originator, TbMsgMetaData metaData, TbMsgDataType dataType, String data) { - return new TbMsg(UUID.randomUUID(), System.currentTimeMillis(), type, originator, metaData.copy(), dataType, data, null, null, TbMsgCallback.EMPTY); + return new TbMsg(ServiceQueue.MAIN, UUID.randomUUID(), System.currentTimeMillis(), type, originator, metaData.copy(), dataType, data, null, null, TbMsgCallback.EMPTY); } public static TbMsg newMsg(String type, EntityId originator, TbMsgMetaData metaData, TbMsgDataType dataType, String data, RuleChainId ruleChainId, RuleNodeId ruleNodeId) { - return new TbMsg(UUID.randomUUID(), System.currentTimeMillis(), type, originator, metaData.copy(), dataType, data, ruleChainId, ruleNodeId, TbMsgCallback.EMPTY); + return new TbMsg(ServiceQueue.MAIN, UUID.randomUUID(), System.currentTimeMillis(), type, originator, metaData.copy(), dataType, data, ruleChainId, ruleNodeId, TbMsgCallback.EMPTY); } public static TbMsg newMsg(String type, EntityId originator, TbMsgMetaData metaData, String data, TbMsgCallback callback) { - return new TbMsg(UUID.randomUUID(), System.currentTimeMillis(), type, originator, metaData.copy(), TbMsgDataType.JSON, data, null, null, callback); + return new TbMsg(ServiceQueue.MAIN, UUID.randomUUID(), System.currentTimeMillis(), type, originator, metaData.copy(), TbMsgDataType.JSON, data, null, null, callback); } public static TbMsg transformMsg(TbMsg origMsg, String type, EntityId originator, TbMsgMetaData metaData, String data) { - return new TbMsg(origMsg.getId(), origMsg.getTs(), type, originator, metaData.copy(), origMsg.getDataType(), + return new TbMsg(origMsg.getQueueName(), origMsg.getId(), origMsg.getTs(), type, originator, metaData.copy(), origMsg.getDataType(), data, origMsg.getRuleChainId(), origMsg.getRuleNodeId(), origMsg.getCallback()); } public static TbMsg newMsg(TbMsg tbMsg, RuleChainId ruleChainId, RuleNodeId ruleNodeId) { - return new TbMsg(UUID.randomUUID(), tbMsg.getTs(), tbMsg.getType(), tbMsg.getOriginator(), tbMsg.getMetaData().copy(), + return new TbMsg(tbMsg.getQueueName(), UUID.randomUUID(), tbMsg.getTs(), tbMsg.getType(), tbMsg.getOriginator(), tbMsg.getMetaData().copy(), tbMsg.getDataType(), tbMsg.getData(), ruleChainId, ruleNodeId, TbMsgCallback.EMPTY); } - private TbMsg(UUID id, long ts, String type, EntityId originator, TbMsgMetaData metaData, TbMsgDataType dataType, String data, + private TbMsg(String queueName, UUID id, long ts, String type, EntityId originator, TbMsgMetaData metaData, TbMsgDataType dataType, String data, RuleChainId ruleChainId, RuleNodeId ruleNodeId, TbMsgCallback callback) { this.id = id; + this.queueName = queueName; if (ts > 0) { this.ts = ts; } else { @@ -136,7 +143,7 @@ public final class TbMsg implements Serializable { return builder.build().toByteArray(); } - public static TbMsg fromBytes(byte[] data, TbMsgCallback callback) { + public static TbMsg fromBytes(String queueName, byte[] data, TbMsgCallback callback) { try { MsgProtos.TbMsgProto proto = MsgProtos.TbMsgProto.parseFrom(data); TbMsgMetaData metaData = new TbMsgMetaData(proto.getMetaData().getDataMap()); @@ -150,18 +157,18 @@ public final class TbMsg implements Serializable { ruleNodeId = new RuleNodeId(new UUID(proto.getRuleNodeIdMSB(), proto.getRuleNodeIdLSB())); } TbMsgDataType dataType = TbMsgDataType.values()[proto.getDataType()]; - return new TbMsg(UUID.fromString(proto.getId()), proto.getTs(), proto.getType(), entityId, metaData, dataType, proto.getData(), ruleChainId, ruleNodeId, callback); + return new TbMsg(queueName, UUID.fromString(proto.getId()), proto.getTs(), proto.getType(), entityId, metaData, dataType, proto.getData(), ruleChainId, ruleNodeId, callback); } catch (InvalidProtocolBufferException e) { throw new IllegalStateException("Could not parse protobuf for TbMsg", e); } } public TbMsg copyWithRuleChainId(RuleChainId ruleChainId) { - return new TbMsg(this.id, this.ts, this.type, this.originator, this.metaData, this.dataType, this.data, ruleChainId, null, callback); + return new TbMsg(this.queueName, this.id, this.ts, this.type, this.originator, this.metaData, this.dataType, this.data, ruleChainId, null, callback); } public TbMsg copyWithRuleNodeId(RuleChainId ruleChainId, RuleNodeId ruleNodeId) { - return new TbMsg(this.id, this.ts, this.type, this.originator, this.metaData, this.dataType, this.data, ruleChainId, ruleNodeId, callback); + return new TbMsg(this.queueName, this.id, this.ts, this.type, this.originator, this.metaData, this.dataType, this.data, ruleChainId, ruleNodeId, callback); } public TbMsgCallback getCallback() { @@ -172,4 +179,8 @@ public final class TbMsg implements Serializable { return TbMsgCallback.EMPTY; } } + + public String getQueueName() { + return queueName != null ? queueName : ServiceQueue.MAIN; + } } diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/queue/ServiceQueue.java b/common/message/src/main/java/org/thingsboard/server/common/msg/queue/ServiceQueue.java index be08f054b6..12abdde8bf 100644 --- a/common/message/src/main/java/org/thingsboard/server/common/msg/queue/ServiceQueue.java +++ b/common/message/src/main/java/org/thingsboard/server/common/msg/queue/ServiceQueue.java @@ -34,7 +34,7 @@ public class ServiceQueue { public ServiceQueue(ServiceType type, String queue) { this.type = type; - this.queue = queue; + this.queue = queue != null ? queue : MAIN; } public ServiceType getType() { diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java index ac19e72037..d62ab6a7ee 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java @@ -521,11 +521,22 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement .setDeviceName(msg.getDeviceInfo().getDeviceName()) .setDeviceType(msg.getDeviceInfo().getDeviceType()) .build(); - transportService.process(sessionInfo, DefaultTransportService.getSessionEventMsg(SessionEvent.OPEN), null); - transportService.registerAsyncSession(sessionInfo, this); - checkGatewaySession(); - ctx.writeAndFlush(createMqttConnAckMsg(CONNECTION_ACCEPTED)); - log.info("[{}] Client connected!", sessionId); + transportService.process(sessionInfo, DefaultTransportService.getSessionEventMsg(SessionEvent.OPEN), new TransportServiceCallback() { + @Override + public void onSuccess(Void msg) { + transportService.registerAsyncSession(sessionInfo, MqttTransportHandler.this); + checkGatewaySession(); + ctx.writeAndFlush(createMqttConnAckMsg(CONNECTION_ACCEPTED)); + log.info("[{}] Client connected!", sessionId); + } + + @Override + public void onError(Throwable e) { + log.warn("[{}] Failed to submit session event", sessionId, e); + ctx.writeAndFlush(createMqttConnAckMsg(MqttConnectReturnCode.CONNECTION_REFUSED_SERVER_UNAVAILABLE)); + ctx.close(); + } + }); } } diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java index a7dac6b63c..aae0ab856e 100644 --- a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java +++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java @@ -130,7 +130,7 @@ public interface TbContext { void ack(TbMsg tbMsg); - TbMsg newMsg(String type, EntityId originator, TbMsgMetaData metaData, String data); + TbMsg newMsg(String queueName, String type, EntityId originator, TbMsgMetaData metaData, String data); TbMsg transformMsg(TbMsg origMsg, String type, EntityId originator, TbMsgMetaData metaData, String data); diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNode.java index c5dc5cc5f4..d56ab0f5e9 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNode.java @@ -136,7 +136,7 @@ public class TbCopyAttributesToEntityViewNode implements TbNode { } private void transformAndTellNext(TbContext ctx, TbMsg msg, EntityView entityView) { - ctx.enqueueForTellNext(ctx.newMsg(msg.getType(), entityView.getId(), msg.getMetaData(), msg.getData()), SUCCESS); + ctx.enqueueForTellNext(ctx.newMsg(msg.getQueueName(), msg.getType(), entityView.getId(), msg.getMetaData(), msg.getData()), SUCCESS); } private boolean attributeContainsInEntityView(String scope, String attrKey, EntityView entityView) { diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbMsgCountNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbMsgCountNode.java index a8f05ae1c4..d2daf9870c 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbMsgCountNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbMsgCountNode.java @@ -24,6 +24,7 @@ import org.thingsboard.server.common.data.plugin.ComponentType; import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsgDataType; import org.thingsboard.server.common.msg.TbMsgMetaData; +import org.thingsboard.server.common.msg.queue.ServiceQueue; import org.thingsboard.server.common.msg.session.SessionMsgType; import java.util.UUID; @@ -74,7 +75,7 @@ public class TbMsgCountNode implements TbNode { TbMsgMetaData metaData = new TbMsgMetaData(); metaData.putValue("delta", Long.toString(System.currentTimeMillis() - lastScheduledTs + delay)); - TbMsg tbMsg = TbMsg.newMsg(SessionMsgType.POST_TELEMETRY_REQUEST.name(), ctx.getTenantId(), metaData, gson.toJson(telemetryJson)); + TbMsg tbMsg = TbMsg.newMsg(msg.getQueueName(), SessionMsgType.POST_TELEMETRY_REQUEST.name(), ctx.getTenantId(), metaData, gson.toJson(telemetryJson)); ctx.enqueueForTellNext(tbMsg, SUCCESS); scheduleTickMsg(ctx); } else { @@ -90,7 +91,7 @@ public class TbMsgCountNode implements TbNode { } lastScheduledTs = lastScheduledTs + delay; long curDelay = Math.max(0L, (lastScheduledTs - curTs)); - TbMsg tickMsg = ctx.newMsg(TB_MSG_COUNT_NODE_MSG, ctx.getSelfId(), new TbMsgMetaData(), ""); + TbMsg tickMsg = ctx.newMsg(ServiceQueue.MAIN, TB_MSG_COUNT_NODE_MSG, ctx.getSelfId(), new TbMsgMetaData(), ""); nextTickId = tickMsg.getId(); ctx.tellSelf(tickMsg, curDelay); } diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/debug/TbMsgGeneratorNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/debug/TbMsgGeneratorNode.java index 46ef675ce1..20e0fa9881 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/debug/TbMsgGeneratorNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/debug/TbMsgGeneratorNode.java @@ -26,6 +26,7 @@ import org.thingsboard.server.common.data.plugin.ComponentType; import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsgMetaData; import org.thingsboard.server.common.msg.queue.PartitionChangeMsg; +import org.thingsboard.server.common.msg.queue.ServiceQueue; import java.util.UUID; import java.util.concurrent.TimeUnit; @@ -118,7 +119,7 @@ public class TbMsgGeneratorNode implements TbNode { } lastScheduledTs = lastScheduledTs + delay; long curDelay = Math.max(0L, (lastScheduledTs - curTs)); - TbMsg tickMsg = ctx.newMsg(TB_MSG_GENERATOR_NODE_MSG, ctx.getSelfId(), new TbMsgMetaData(), ""); + TbMsg tickMsg = ctx.newMsg(ServiceQueue.MAIN, TB_MSG_GENERATOR_NODE_MSG, ctx.getSelfId(), new TbMsgMetaData(), ""); nextTickId = tickMsg.getId(); ctx.tellSelf(tickMsg, curDelay); } @@ -126,13 +127,13 @@ public class TbMsgGeneratorNode implements TbNode { private ListenableFuture generate(TbContext ctx) { return ctx.getJsExecutor().executeAsync(() -> { if (prevMsg == null) { - prevMsg = ctx.newMsg("", originatorId, new TbMsgMetaData(), "{}"); + prevMsg = ctx.newMsg(ServiceQueue.MAIN, "", originatorId, new TbMsgMetaData(), "{}"); } if (initialized) { ctx.logJsEvalRequest(); TbMsg generated = jsEngine.executeGenerate(prevMsg); ctx.logJsEvalResponse(); - prevMsg = ctx.newMsg(generated.getType(), originatorId, generated.getMetaData(), generated.getData()); + prevMsg = ctx.newMsg(ServiceQueue.MAIN, generated.getType(), originatorId, generated.getMetaData(), generated.getData()); } return prevMsg; }); diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/delay/TbMsgDelayNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/delay/TbMsgDelayNode.java index e2bb07fb6f..45e3a2b2aa 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/delay/TbMsgDelayNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/delay/TbMsgDelayNode.java @@ -26,6 +26,7 @@ import org.thingsboard.rule.engine.api.util.TbNodeUtils; import org.thingsboard.server.common.data.plugin.ComponentType; import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsgMetaData; +import org.thingsboard.server.common.msg.queue.ServiceQueue; import java.util.HashMap; import java.util.Map; @@ -70,7 +71,7 @@ public class TbMsgDelayNode implements TbNode { } else { if (pendingMsgs.size() < config.getMaxPendingMsgs()) { pendingMsgs.put(msg.getId(), msg); - TbMsg tickMsg = ctx.newMsg(TB_MSG_DELAY_NODE_MSG, ctx.getSelfId(), new TbMsgMetaData(), msg.getId().toString()); + TbMsg tickMsg = ctx.newMsg(ServiceQueue.MAIN, TB_MSG_DELAY_NODE_MSG, ctx.getSelfId(), new TbMsgMetaData(), msg.getId().toString()); ctx.tellSelf(tickMsg, getDelay(msg)); ctx.ack(msg); } else { diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rpc/TbSendRPCRequestNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rpc/TbSendRPCRequestNode.java index eb441ff99e..9f2c35b7ad 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rpc/TbSendRPCRequestNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rpc/TbSendRPCRequestNode.java @@ -112,11 +112,11 @@ public class TbSendRPCRequestNode implements TbNode { ctx.getRpcService().sendRpcRequestToDevice(request, ruleEngineDeviceRpcResponse -> { if (!ruleEngineDeviceRpcResponse.getError().isPresent()) { - TbMsg next = ctx.newMsg(msg.getType(), msg.getOriginator(), msg.getMetaData(), ruleEngineDeviceRpcResponse.getResponse().orElse("{}")); + TbMsg next = ctx.newMsg(msg.getQueueName(), msg.getType(), msg.getOriginator(), msg.getMetaData(), ruleEngineDeviceRpcResponse.getResponse().orElse("{}")); ctx.enqueueForTellNext(next, TbRelationTypes.SUCCESS); } else { - TbMsg next = ctx.newMsg(msg.getType(), msg.getOriginator(), msg.getMetaData(), wrap("error", ruleEngineDeviceRpcResponse.getError().get().name())); - ctx.enqueueForTellFailure(next, ruleEngineDeviceRpcResponse.getError().get().name()); + TbMsg next = ctx.newMsg(msg.getQueueName(), msg.getType(), msg.getOriginator(), msg.getMetaData(), wrap("error", ruleEngineDeviceRpcResponse.getError().get().name())); + ctx.tellFailure(next, new RuntimeException(ruleEngineDeviceRpcResponse.getError().get().name())); } }); ctx.ack(msg);