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/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(); + } + }); } }