diff --git a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java index 26c82a33de..61d9586095 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java +++ b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java @@ -32,6 +32,7 @@ import org.springframework.stereotype.Component; import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.rule.engine.api.DeviceStateManager; import org.thingsboard.rule.engine.api.MailService; +import org.thingsboard.rule.engine.api.MqttClientSettings; import org.thingsboard.rule.engine.api.NotificationCenter; import org.thingsboard.rule.engine.api.SmsService; import org.thingsboard.rule.engine.api.notification.SlackService; @@ -639,6 +640,10 @@ public class ActorSystemContext { @Getter private long cfCalculationResultTimeout; + @Autowired + @Getter + private MqttClientSettings mqttClientSettings; + @Getter @Setter private TbActorSystem actorSystem; 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 033e10ca9a..3fb28aee38 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 @@ -23,14 +23,15 @@ import org.bouncycastle.util.Arrays; import org.thingsboard.common.util.DebugModeUtil; import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.common.util.ListeningExecutor; +import org.thingsboard.rule.engine.api.DeviceStateManager; import org.thingsboard.rule.engine.api.MailService; +import org.thingsboard.rule.engine.api.MqttClientSettings; import org.thingsboard.rule.engine.api.NotificationCenter; import org.thingsboard.rule.engine.api.RuleEngineAlarmService; import org.thingsboard.rule.engine.api.RuleEngineApiUsageStateService; import org.thingsboard.rule.engine.api.RuleEngineAssetProfileCache; import org.thingsboard.rule.engine.api.RuleEngineCalculatedFieldQueueService; import org.thingsboard.rule.engine.api.RuleEngineDeviceProfileCache; -import org.thingsboard.rule.engine.api.DeviceStateManager; import org.thingsboard.rule.engine.api.RuleEngineRpcService; import org.thingsboard.rule.engine.api.RuleEngineTelemetryService; import org.thingsboard.rule.engine.api.ScriptEngine; @@ -1010,13 +1011,17 @@ public class DefaultTbContext implements TbContext { return mainCtx.getAuditLogService(); } + @Override + public MqttClientSettings getMqttClientSettings() { + return mainCtx.getMqttClientSettings(); + } + private TbMsgMetaData getActionMetaData(RuleNodeId ruleNodeId) { TbMsgMetaData metaData = new TbMsgMetaData(); metaData.putValue("ruleNodeId", ruleNodeId.toString()); return metaData; } - @Override public void schedule(Runnable runnable, long delay, TimeUnit timeUnit) { mainCtx.getScheduler().schedule(runnable, delay, timeUnit); diff --git a/application/src/main/java/org/thingsboard/server/config/mqtt/MqttClientRetransmissionSettingsComponent.java b/application/src/main/java/org/thingsboard/server/config/mqtt/MqttClientRetransmissionSettingsComponent.java new file mode 100644 index 0000000000..33e9358d2b --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/config/mqtt/MqttClientRetransmissionSettingsComponent.java @@ -0,0 +1,37 @@ +/** + * Copyright © 2016-2025 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.config.mqtt; + +import jakarta.validation.constraints.PositiveOrZero; +import lombok.Data; +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.context.annotation.Configuration; +import org.springframework.validation.annotation.Validated; + +@Data +@Validated +@Configuration +@ConfigurationProperties(prefix = "mqtt.client.retransmission") +public class MqttClientRetransmissionSettingsComponent { + + @PositiveOrZero + private int maxAttempts; + @PositiveOrZero + private long initialDelayMillis; + @PositiveOrZero + private double jitterFactor; + +} diff --git a/application/src/main/java/org/thingsboard/server/config/mqtt/MqttClientSettingsComponent.java b/application/src/main/java/org/thingsboard/server/config/mqtt/MqttClientSettingsComponent.java new file mode 100644 index 0000000000..25df212925 --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/config/mqtt/MqttClientSettingsComponent.java @@ -0,0 +1,47 @@ +/** + * Copyright © 2016-2025 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.config.mqtt; + +import lombok.EqualsAndHashCode; +import lombok.RequiredArgsConstructor; +import lombok.ToString; +import org.springframework.context.annotation.Configuration; +import org.thingsboard.rule.engine.api.MqttClientSettings; + +@ToString +@EqualsAndHashCode +@Configuration +@RequiredArgsConstructor +public class MqttClientSettingsComponent implements MqttClientSettings { + + private final MqttClientRetransmissionSettingsComponent retransmissionSettingsComponent; + + @Override + public int getRetransmissionMaxAttempts() { + return retransmissionSettingsComponent.getMaxAttempts(); + } + + @Override + public long getRetransmissionInitialDelayMillis() { + return retransmissionSettingsComponent.getInitialDelayMillis(); + } + + @Override + public double getRetransmissionJitterFactor() { + return retransmissionSettingsComponent.getJitterFactor(); + } + +} diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index d1910399f0..df33a92857 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -1952,3 +1952,27 @@ mobileApp: googlePlayLink: "${TB_MOBILE_APP_GOOGLE_PLAY_LINK:https://play.google.com/store/apps/details?id=org.thingsboard.demo.app}" # Link to App Store for Thingsboard Live mobile application appStoreLink: "${TB_MOBILE_APP_APP_STORE_LINK:https://apps.apple.com/us/app/thingsboard-live/id1594355695}" + +mqtt: + # MQTT client configuration parameters + client: + # Parameters that control the retransmission mechanism. + # This mechanism only applies to the handling of MQTT Publish, Subscribe, Unsubscribe and Pubrel messages. + # With the updated default settings: + # - After sending the message, wait approximately 5000 ms (± jitter) for the 1st attempt. + # - The 2nd attempt will occur after roughly 5000 * 2 = 10,000 ms (± jitter). + # - The 3rd attempt will occur after roughly 5000 * 4 = 20,000 ms (± jitter). + # - The 4th "attempt" will not actually perform a retransmission. + # Instead, the system will detect that the maximum number of attempts has been reached and drop the pending message. + retransmission: + # Maximum number of retransmission attempts allowed. + # If the attempt count exceeds this value, retransmissions will stop and the pending message will be dropped. + max_attempts: "${TB_MQTT_CLIENT_RETRANSMISSION_MAX_ATTEMPTS:3}" + # Base delay (in milliseconds) before the first retransmission attempt, measured from the moment the message is sent. + # Subsequent delays are calculated using exponential backoff. + # This base delay is also used as the reference value for applying jitter. + initial_delay_millis: "${TB_MQTT_CLIENT_RETRANSMISSION_INITIAL_DELAY_MILLIS:5000}" + # Jitter factor applied to the calculated retransmission delay. + # The actual delay is randomized within a range defined by multiplying the base delay by a factor between (1 - jitter_factor) and (1 + jitter_factor). + # For example, a jitter_factor of 0.15 means the actual delay may vary by up to ±15% of the base delay. + jitter_factor: "${TB_MQTT_CLIENT_RETRANSMISSION_JITTER_FACTOR:0.15}" diff --git a/msa/black-box-tests/src/test/java/org/thingsboard/server/msa/ContainerTestSuite.java b/msa/black-box-tests/src/test/java/org/thingsboard/server/msa/ContainerTestSuite.java index f2e721bdcb..e00b0828c9 100644 --- a/msa/black-box-tests/src/test/java/org/thingsboard/server/msa/ContainerTestSuite.java +++ b/msa/black-box-tests/src/test/java/org/thingsboard/server/msa/ContainerTestSuite.java @@ -55,8 +55,8 @@ public class ContainerTestSuite { private static final String TB_JS_EXECUTOR_LOG_REGEXP = ".*template started.*"; private static final Duration CONTAINER_STARTUP_TIMEOUT = Duration.ofSeconds(400); - private DockerComposeContainer testContainer; - private ThingsBoardDbInstaller installTb; + private DockerComposeContainer testContainer; + private ThingsBoardDbInstaller installTb; private boolean isActive; private static ContainerTestSuite containerTestSuite; @@ -194,7 +194,7 @@ public class ContainerTestSuite { setActive(true); } catch (Exception e) { log.error("Failed to create test container", e); - fail("Failed to create test container"); + fail("Failed to create test container", e); } } @@ -263,7 +263,7 @@ public class ContainerTestSuite { log.info("Trying to delete temp dir {}", targetDir); FileUtils.deleteDirectory(new File(targetDir)); } catch (IOException e) { - log.error("Can't delete temp directory " + targetDir, e); + log.error("Can't delete temp directory {}", targetDir, e); } } @@ -286,8 +286,8 @@ public class ContainerTestSuite { FileUtils.writeStringToFile(file, outputContent, StandardCharsets.UTF_8); assertThat(FileUtils.readFileToString(file, StandardCharsets.UTF_8), is(outputContent)); } catch (IOException e) { - log.error("failed to update file " + sourceFilename, e); - fail("failed to update file"); + log.error("failed to update file {}", sourceFilename, e); + fail("failed to update file", e); } } diff --git a/msa/black-box-tests/src/test/java/org/thingsboard/server/msa/connectivity/MqttClientTest.java b/msa/black-box-tests/src/test/java/org/thingsboard/server/msa/connectivity/MqttClientTest.java index ebdfb4e3c9..1b1ed9bf0f 100644 --- a/msa/black-box-tests/src/test/java/org/thingsboard/server/msa/connectivity/MqttClientTest.java +++ b/msa/black-box-tests/src/test/java/org/thingsboard/server/msa/connectivity/MqttClientTest.java @@ -42,7 +42,6 @@ import org.thingsboard.mqtt.MqttClient; import org.thingsboard.mqtt.MqttClientCallback; import org.thingsboard.mqtt.MqttClientConfig; import org.thingsboard.mqtt.MqttHandler; -import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.DeviceProfile; import org.thingsboard.server.common.data.DeviceProfileProvisionType; @@ -82,7 +81,6 @@ import java.util.concurrent.TimeoutException; import static org.assertj.core.api.Assertions.assertThat; import static org.testng.Assert.assertNotNull; import static org.testng.Assert.fail; -import static org.thingsboard.server.common.data.DataConstants.DEVICE; import static org.thingsboard.server.common.data.DataConstants.SHARED_SCOPE; import static org.thingsboard.server.msa.prototypes.DevicePrototypes.defaultDevicePrototype; @@ -301,7 +299,7 @@ public class MqttClientTest extends AbstractContainerTest { assertThat(Objects.requireNonNull(requestFromServer).getMessage()).isEqualTo("{\"method\":\"getValue\",\"params\":true}"); - Integer requestId = Integer.valueOf(Objects.requireNonNull(requestFromServer).getTopic().substring("v1/devices/me/rpc/request/".length())); + int requestId = Integer.parseInt(Objects.requireNonNull(requestFromServer).getTopic().substring("v1/devices/me/rpc/request/".length())); JsonObject clientResponse = new JsonObject(); clientResponse.addProperty("response", "someResponse"); // Send a response to the server's RPC request @@ -340,7 +338,7 @@ public class MqttClientTest extends AbstractContainerTest { assertThat(Objects.requireNonNull(requestFromServer).getMessage()).isEqualTo("{\"method\":\"getValue\",\"params\":true}"); - Integer requestId = Integer.valueOf(Objects.requireNonNull(requestFromServer).getTopic().substring("v1/devices/me/rpc/request/".length())); + int requestId = Integer.parseInt(Objects.requireNonNull(requestFromServer).getTopic().substring("v1/devices/me/rpc/request/".length())); JsonObject clientResponse = new JsonObject(); clientResponse.addProperty("response", "someResponse"); // Send a response to the server's RPC request @@ -520,13 +518,13 @@ public class MqttClientTest extends AbstractContainerTest { mqttClient.on("/provision/response", listener, MqttQoS.AT_LEAST_ONCE).get(3 * timeoutMultiplier, TimeUnit.SECONDS); TimeUnit.SECONDS.sleep(2 * timeoutMultiplier); assertThat(subAckResult[0]).isNotNull(); - assertThat(MqttReasonCodes.SubAck.GRANTED_QOS_1.equals(subAckResult[0])); + assertThat(MqttReasonCodes.SubAck.GRANTED_QOS_1).isEqualTo(subAckResult[0]); subAckResult[0] = null; mqttClient.on("v1/devices/me/attributes", listener, MqttQoS.AT_LEAST_ONCE).get(3 * timeoutMultiplier, TimeUnit.SECONDS); TimeUnit.SECONDS.sleep(2 * timeoutMultiplier); assertThat(subAckResult[0]).isNotNull(); - assertThat(MqttReasonCodes.SubAck.TOPIC_FILTER_INVALID.equals(subAckResult[0])); + assertThat(MqttReasonCodes.SubAck.TOPIC_FILTER_INVALID).isEqualTo(subAckResult[0]); testRestClient.deleteDeviceIfExists(device.getId()); updateDeviceProfileWithProvisioningStrategy(deviceProfile, DeviceProfileProvisionType.DISABLED); @@ -596,7 +594,7 @@ public class MqttClientTest extends AbstractContainerTest { .await() .alias("Check device disconnect.") .atMost(TIMEOUT*timeoutMultiplier, TimeUnit.SECONDS) - .until(() -> returnCodeByteValue.size() > 0); + .until(() -> !returnCodeByteValue.isEmpty()); assertThat(returnCodeByteValueSecondClient).isEmpty(); assertThat(returnCodeByteValue).isNotEmpty(); @@ -663,7 +661,7 @@ public class MqttClientTest extends AbstractContainerTest { .stream() .filter(RuleChain::isRoot) .findFirst(); - if (!defaultRuleChain.isPresent()) { + if (defaultRuleChain.isEmpty()) { fail("Root rule chain wasn't found"); } return defaultRuleChain.get().getId(); @@ -717,6 +715,7 @@ public class MqttClientTest extends AbstractContainerTest { clientConfig.setClientId("MQTT client from test"); clientConfig.setUsername(username); clientConfig.setProtocolVersion(mqttVersion); + clientConfig.setRetransmissionConfig(new MqttClientConfig.RetransmissionConfig(3, 5000L, 0.15d)); // same as defaults in thingsboard.yml as of time of this writing MqttClient mqttClient = MqttClient.create(clientConfig, listener, handlerExecutor); if (connect) { mqttClient.connect(TRANSPORT_HOST, TRANSPORT_PORT).get(); diff --git a/msa/black-box-tests/src/test/java/org/thingsboard/server/msa/connectivity/MqttGatewayClientTest.java b/msa/black-box-tests/src/test/java/org/thingsboard/server/msa/connectivity/MqttGatewayClientTest.java index cc587fbbd5..fd9d4f557d 100644 --- a/msa/black-box-tests/src/test/java/org/thingsboard/server/msa/connectivity/MqttGatewayClientTest.java +++ b/msa/black-box-tests/src/test/java/org/thingsboard/server/msa/connectivity/MqttGatewayClientTest.java @@ -39,7 +39,6 @@ import org.thingsboard.common.util.ThingsBoardThreadFactory; import org.thingsboard.mqtt.MqttClient; import org.thingsboard.mqtt.MqttClientConfig; import org.thingsboard.mqtt.MqttHandler; -import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.StringUtils; import org.thingsboard.server.common.data.id.DeviceId; @@ -65,7 +64,6 @@ import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import static org.assertj.core.api.Assertions.assertThat; -import static org.thingsboard.server.common.data.DataConstants.DEVICE; import static org.thingsboard.server.common.data.DataConstants.SHARED_SCOPE; import static org.thingsboard.server.msa.prototypes.DevicePrototypes.defaultGatewayPrototype; @@ -76,7 +74,6 @@ public class MqttGatewayClientTest extends AbstractContainerTest { private MqttClient mqttClient; private Device createdDevice; private MqttMessageListener listener; - private JsonParser jsonParser = new JsonParser(); AbstractListeningExecutor handlerExecutor; @@ -100,7 +97,7 @@ public class MqttGatewayClientTest extends AbstractContainerTest { } @AfterMethod - public void removeGateway() { + public void removeGateway() { testRestClient.deleteDeviceIfExists(this.gatewayDevice.getId()); testRestClient.deleteDeviceIfExists(this.createdDevice.getId()); this.listener = null; @@ -197,7 +194,7 @@ public class MqttGatewayClientTest extends AbstractContainerTest { mqttClient.publish("v1/gateway/attributes/request", Unpooled.wrappedBuffer(requestData.toString().getBytes())).get(); event = listener.getEvents().poll(10 * timeoutMultiplier, TimeUnit.SECONDS); - JsonObject responseData = jsonParser.parse(Objects.requireNonNull(event).getMessage()).getAsJsonObject(); + JsonObject responseData = JsonParser.parseString(Objects.requireNonNull(event).getMessage()).getAsJsonObject(); assertThat(responseData.has("value")).isTrue(); assertThat(responseData.get("value").getAsString()).isEqualTo(sharedAttributes.get("attr1").getAsString()); @@ -213,7 +210,7 @@ public class MqttGatewayClientTest extends AbstractContainerTest { mqttClient.on("v1/gateway/attributes/response", listener, MqttQoS.AT_LEAST_ONCE).get(); mqttClient.publish("v1/gateway/attributes/request", Unpooled.wrappedBuffer(requestData.toString().getBytes())).get(); event = listener.getEvents().poll(10 * timeoutMultiplier, TimeUnit.SECONDS); - responseData = jsonParser.parse(Objects.requireNonNull(event).getMessage()).getAsJsonObject(); + responseData = JsonParser.parseString(Objects.requireNonNull(event).getMessage()).getAsJsonObject(); assertThat(responseData.has("values")).isTrue(); assertThat(responseData.get("values").getAsJsonObject().get("attr1").getAsString()).isEqualTo(sharedAttributes.get("attr1").getAsString()); @@ -231,7 +228,7 @@ public class MqttGatewayClientTest extends AbstractContainerTest { mqttClient.on("v1/gateway/attributes/response", listener, MqttQoS.AT_LEAST_ONCE).get(); mqttClient.publish("v1/gateway/attributes/request", Unpooled.wrappedBuffer(requestData.toString().getBytes())).get(); event = listener.getEvents().poll(10 * timeoutMultiplier, TimeUnit.SECONDS); - responseData = jsonParser.parse(Objects.requireNonNull(event).getMessage()).getAsJsonObject(); + responseData = JsonParser.parseString(Objects.requireNonNull(event).getMessage()).getAsJsonObject(); assertThat(responseData.has("values")).isTrue(); assertThat(responseData.get("values").getAsJsonObject().get("attr1").getAsString()).isEqualTo(sharedAttributes.get("attr1").getAsString()); @@ -390,7 +387,7 @@ public class MqttGatewayClientTest extends AbstractContainerTest { mqttClient.publish("v1/gateway/attributes/request", Unpooled.wrappedBuffer(gatewayAttributesRequest.toString().getBytes())).get(); MqttEvent clientAttributeEvent = listener.getEvents().poll(10 * timeoutMultiplier, TimeUnit.SECONDS); assertThat(clientAttributeEvent).isNotNull(); - JsonObject responseMessage = new JsonParser().parse(Objects.requireNonNull(clientAttributeEvent).getMessage()).getAsJsonObject(); + JsonObject responseMessage = JsonParser.parseString(Objects.requireNonNull(clientAttributeEvent).getMessage()).getAsJsonObject(); assertThat(responseMessage.get("id").getAsInt()).isEqualTo(messageId); assertThat(responseMessage.get("device").getAsString()).isEqualTo(createdDevice.getName()); @@ -427,6 +424,7 @@ public class MqttGatewayClientTest extends AbstractContainerTest { clientConfig.setOwnerId(getOwnerId()); clientConfig.setClientId("MQTT client from test"); clientConfig.setUsername(deviceCredentials.getCredentialsId()); + clientConfig.setRetransmissionConfig(new MqttClientConfig.RetransmissionConfig(3, 5000L, 0.15d)); // same as defaults in thingsboard.yml as of time of this writing MqttClient mqttClient = MqttClient.create(clientConfig, listener, handlerExecutor); mqttClient.connect("localhost", 1883).get(); return mqttClient; diff --git a/netty-mqtt/pom.xml b/netty-mqtt/pom.xml index 7e0a00eee6..8ce2c68310 100644 --- a/netty-mqtt/pom.xml +++ b/netty-mqtt/pom.xml @@ -87,6 +87,26 @@ awaitility test + + org.testcontainers + testcontainers + test + + + org.testcontainers + junit-jupiter + test + + + software.xdev + testcontainers-junit4-mock + test + + + org.testcontainers + hivemq + test + diff --git a/netty-mqtt/src/main/java/org/thingsboard/mqtt/MaxRetransmissionsReachedException.java b/netty-mqtt/src/main/java/org/thingsboard/mqtt/MaxRetransmissionsReachedException.java new file mode 100644 index 0000000000..3d483dd541 --- /dev/null +++ b/netty-mqtt/src/main/java/org/thingsboard/mqtt/MaxRetransmissionsReachedException.java @@ -0,0 +1,24 @@ +/** + * Copyright © 2016-2025 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.mqtt; + +public class MaxRetransmissionsReachedException extends RuntimeException { + + public MaxRetransmissionsReachedException(String message) { + super(message); + } + +} diff --git a/netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttChannelHandler.java b/netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttChannelHandler.java index ad976c848a..9686a2b1d7 100644 --- a/netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttChannelHandler.java +++ b/netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttChannelHandler.java @@ -57,7 +57,7 @@ final class MqttChannelHandler extends SimpleChannelInboundHandler } @Override - protected void channelRead0(ChannelHandlerContext ctx, MqttMessage msg) throws Exception { + protected void channelRead0(ChannelHandlerContext ctx, MqttMessage msg) { if (msg.decoderResult().isSuccess()) { switch (msg.fixedHeader().messageType()) { case CONNACK: @@ -120,6 +120,7 @@ final class MqttChannelHandler extends SimpleChannelInboundHandler this.client.getClientConfig().getUsername(), this.client.getClientConfig().getPassword() != null ? this.client.getClientConfig().getPassword().getBytes(CharsetUtil.UTF_8) : null ); + log.debug("{} Sending CONNECT", client.getClientConfig().getOwnerId()); ctx.channel().writeAndFlush(new MqttConnectMessage(fixedHeader, variableHeader, payload)); } @@ -173,6 +174,7 @@ final class MqttChannelHandler extends SimpleChannelInboundHandler } private void handleConack(Channel channel, MqttConnAckMessage message) { + log.debug("{} Handling CONNACK", client.getClientConfig().getOwnerId()); switch (message.variableHeader().connectReturnCode()) { case CONNECTION_ACCEPTED: this.connectFuture.setSuccess(new MqttConnectResult(true, MqttConnectReturnCode.CONNECTION_ACCEPTED, channel.closeFuture())); @@ -219,9 +221,9 @@ final class MqttChannelHandler extends SimpleChannelInboundHandler } pendingSubscription.onSubackReceived(); for (MqttPendingSubscription.MqttPendingHandler handler : pendingSubscription.getHandlers()) { - MqttSubscription subscription = new MqttSubscription(pendingSubscription.getTopic(), handler.getHandler(), handler.isOnce()); + MqttSubscription subscription = new MqttSubscription(pendingSubscription.getTopic(), handler.handler(), handler.once()); this.client.getSubscriptions().put(pendingSubscription.getTopic(), subscription); - this.client.getHandlerToSubscription().put(handler.getHandler(), subscription); + this.client.getHandlerToSubscription().put(handler.handler(), subscription); } this.client.getPendingSubscribeTopics().remove(pendingSubscription.getTopic()); @@ -282,17 +284,16 @@ final class MqttChannelHandler extends SimpleChannelInboundHandler } private void handlePuback(MqttPubAckMessage message) { - MqttPendingPublish pendingPublish = this.client.getPendingPublishes().get(message.variableHeader().messageId()); - if (pendingPublish == null) { - return; - } - pendingPublish.getFuture().setSuccess(null); - pendingPublish.onPubackReceived(); - this.client.getPendingPublishes().remove(message.variableHeader().messageId()); - pendingPublish.getPayload().release(); - if (this.client.getCallback() != null) { - this.client.getCallback().onPubAck(message); - } + log.trace("{} Handling PUBACK", client.getClientConfig().getOwnerId()); + client.getPendingPublishes().computeIfPresent(message.variableHeader().messageId(), (__, pendingPublish) -> { + pendingPublish.getFuture().setSuccess(null); + pendingPublish.onPubackReceived(); + pendingPublish.getPayload().release(); + if (client.getCallback() != null) { + client.getCallback().onPubAck(message); + } + return null; + }); } private void handlePubrec(Channel channel, MqttMessage message) { @@ -335,6 +336,7 @@ final class MqttChannelHandler extends SimpleChannelInboundHandler } private void handleDisconnect(MqttMessage message) { + log.debug("{} Handling DISCONNECT", client.getClientConfig().getOwnerId()); if (this.client.getCallback() != null) { this.client.getCallback().onDisconnect(message); } diff --git a/netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttClient.java b/netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttClient.java index db0459e08a..4d845320e8 100644 --- a/netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttClient.java +++ b/netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttClient.java @@ -184,7 +184,7 @@ public interface MqttClient { * @param config The config object to use while looking for settings * @param defaultHandler The handler for incoming messages that do not match any topic subscriptions */ - static MqttClient create(MqttClientConfig config, MqttHandler defaultHandler, ListeningExecutor handlerExecutor){ + static MqttClient create(MqttClientConfig config, MqttHandler defaultHandler, ListeningExecutor handlerExecutor) { return new MqttClientImpl(config, defaultHandler, handlerExecutor); } diff --git a/netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttClientConfig.java b/netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttClientConfig.java index 41df077d71..24feb3e58e 100644 --- a/netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttClientConfig.java +++ b/netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttClientConfig.java @@ -47,6 +47,26 @@ public final class MqttClientConfig { private long reconnectDelay = 1L; private int maxBytesInMessage = 8092; + @Getter + @Setter + private RetransmissionConfig retransmissionConfig; + + public record RetransmissionConfig(int maxAttempts, long initialDelayMillis, double jitterFactor) { + + public RetransmissionConfig { + if (maxAttempts < 0) { + throw new IllegalArgumentException("Max retransmission attempts (maxAttempts) must be zero or greater, but was " + maxAttempts); + } + if (initialDelayMillis < 0) { + throw new IllegalArgumentException("Initial retransmission delay (initialDelayMillis) must be zero or greater, but was " + initialDelayMillis); + } + if (jitterFactor < 0) { + throw new IllegalArgumentException("Jitter factor (jitterFactor) must be zero or greater, but was " + jitterFactor); + } + } + + } + public MqttClientConfig() { this(null); } diff --git a/netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttClientImpl.java b/netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttClientImpl.java index 47eae565dc..ee07752db3 100644 --- a/netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttClientImpl.java +++ b/netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttClientImpl.java @@ -17,6 +17,7 @@ package org.thingsboard.mqtt; import com.google.common.collect.HashMultimap; import com.google.common.collect.ImmutableSet; +import com.google.common.collect.Sets; import io.netty.bootstrap.Bootstrap; import io.netty.buffer.ByteBuf; import io.netty.channel.Channel; @@ -384,8 +385,33 @@ final class MqttClientImpl implements MqttClient { MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, false, qos, retain, 0); MqttPublishVariableHeader variableHeader = new MqttPublishVariableHeader(topic, getNewMessageId().messageId()); MqttPublishMessage message = new MqttPublishMessage(fixedHeader, variableHeader, payload); - MqttPendingPublish pendingPublish = new MqttPendingPublish(variableHeader.packetId(), future, - payload.retain(), message, qos, () -> !pendingPublishes.containsKey(variableHeader.packetId())); + + final var pendingPublish = MqttPendingPublish.builder() + .messageId(variableHeader.packetId()) + .future(future) + .payload(payload.retain()) + .message(message) + .qos(qos) + .ownerId(clientConfig.getOwnerId()) + .retransmissionConfig(clientConfig.getRetransmissionConfig()) + .pendingOperation(new PendingOperation() { + @Override + public boolean isCancelled() { + return !pendingPublishes.containsKey(variableHeader.packetId()); + } + + @Override + public void onMaxRetransmissionAttemptsReached() { + pendingPublishes.computeIfPresent(variableHeader.packetId(), (__, pendingPublish) -> { + var message = "Unable to deliver publish message due to max retransmission attempts (%s) being reached for client '%s' on topic '%s' (message ID: %d)" + .formatted(clientConfig.getRetransmissionConfig().maxAttempts(), clientConfig.getClientId(), topic, variableHeader.packetId()); + pendingPublish.getFuture().tryFailure(new MaxRetransmissionsReachedException(message)); + pendingPublish.getPayload().release(); + return null; + }); + } + }).build(); + this.pendingPublishes.put(pendingPublish.getMessageId(), pendingPublish); ChannelFuture channelFuture = this.sendAndFlushPacket(message); @@ -499,9 +525,30 @@ final class MqttClientImpl implements MqttClient { MqttSubscribePayload payload = new MqttSubscribePayload(Collections.singletonList(subscription)); MqttSubscribeMessage message = new MqttSubscribeMessage(fixedHeader, variableHeader, payload); - final MqttPendingSubscription pendingSubscription = new MqttPendingSubscription(future, topic, message, - () -> !pendingSubscriptions.containsKey(variableHeader.messageId())); - pendingSubscription.addHandler(handler, once); + final var pendingSubscription = MqttPendingSubscription.builder() + .future(future) + .topic(topic) + .handlers(Sets.newHashSet(new MqttPendingSubscription.MqttPendingHandler(handler, once))) + .subscribeMessage(message) + .ownerId(clientConfig.getOwnerId()) + .retransmissionConfig(clientConfig.getRetransmissionConfig()) + .pendingOperation(new PendingOperation() { + @Override + public boolean isCancelled() { + return !pendingSubscriptions.containsKey(variableHeader.messageId()); + } + + @Override + public void onMaxRetransmissionAttemptsReached() { + pendingSubscriptions.computeIfPresent(variableHeader.messageId(), (__, pendingSubscription) -> { + var message = "Unable to deliver subscribe message due to max retransmission attempts (%s) being reached for client '%s' on topic '%s' (message ID: %d)" + .formatted(clientConfig.getRetransmissionConfig().maxAttempts(), clientConfig.getClientId(), topic, variableHeader.messageId()); + pendingSubscription.getFuture().tryFailure(new MaxRetransmissionsReachedException(message)); + return null; + }); + } + }).build(); + this.pendingSubscriptions.put(variableHeader.messageId(), pendingSubscription); this.pendingSubscribeTopics.add(topic); pendingSubscription.setSent(this.sendAndFlushPacket(message) != null); //If not sent, we will send it when the connection is opened @@ -518,8 +565,29 @@ final class MqttClientImpl implements MqttClient { MqttUnsubscribePayload payload = new MqttUnsubscribePayload(Collections.singletonList(topic)); MqttUnsubscribeMessage message = new MqttUnsubscribeMessage(fixedHeader, variableHeader, payload); - MqttPendingUnsubscription pendingUnsubscription = new MqttPendingUnsubscription(promise, topic, message, - () -> !pendingServerUnsubscribes.containsKey(variableHeader.messageId())); + final var pendingUnsubscription = MqttPendingUnsubscription.builder() + .future(promise) + .topic(topic) + .unsubscribeMessage(message) + .ownerId(clientConfig.getOwnerId()) + .retransmissionConfig(clientConfig.getRetransmissionConfig()) + .pendingOperation(new PendingOperation() { + @Override + public boolean isCancelled() { + return !pendingServerUnsubscribes.containsKey(variableHeader.messageId()); + } + + @Override + public void onMaxRetransmissionAttemptsReached() { + pendingServerUnsubscribes.computeIfPresent(variableHeader.messageId(), (__, pendingUnsubscription) -> { + var message = "Unable to deliver unsubscribe message due to max retransmission attempts (%s) being reached for client '%s' on topic '%s' (message ID: %d)" + .formatted(clientConfig.getRetransmissionConfig().maxAttempts(), clientConfig.getClientId(), topic, variableHeader.messageId()); + pendingUnsubscription.getFuture().tryFailure(new MaxRetransmissionsReachedException(message)); + return null; + }); + } + }).build(); + this.pendingServerUnsubscribes.put(variableHeader.messageId(), pendingUnsubscription); pendingUnsubscription.startRetransmissionTimer(this.eventLoop.next(), this::sendAndFlushPacket); diff --git a/netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttConnectResult.java b/netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttConnectResult.java index 911bc1d395..67757d2a7a 100644 --- a/netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttConnectResult.java +++ b/netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttConnectResult.java @@ -17,7 +17,9 @@ package org.thingsboard.mqtt; import io.netty.channel.ChannelFuture; import io.netty.handler.codec.mqtt.MqttConnectReturnCode; +import lombok.ToString; +@ToString @SuppressWarnings({"WeakerAccess", "unused"}) public final class MqttConnectResult { @@ -42,4 +44,5 @@ public final class MqttConnectResult { public ChannelFuture getCloseFuture() { return closeFuture; } + } diff --git a/netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttPendingPublish.java b/netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttPendingPublish.java index e8c3ef35f7..1846bdb12b 100644 --- a/netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttPendingPublish.java +++ b/netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttPendingPublish.java @@ -21,9 +21,13 @@ import io.netty.handler.codec.mqtt.MqttMessage; import io.netty.handler.codec.mqtt.MqttPublishMessage; import io.netty.handler.codec.mqtt.MqttQoS; import io.netty.util.concurrent.Promise; +import lombok.AccessLevel; +import lombok.Getter; +import lombok.Setter; import java.util.function.Consumer; +@Getter(AccessLevel.PACKAGE) final class MqttPendingPublish { private final int messageId; @@ -32,80 +36,126 @@ final class MqttPendingPublish { private final MqttPublishMessage message; private final MqttQoS qos; + @Getter(AccessLevel.NONE) private final RetransmissionHandler publishRetransmissionHandler; + @Getter(AccessLevel.NONE) private final RetransmissionHandler pubrelRetransmissionHandler; + @Setter(AccessLevel.PACKAGE) private boolean sent = false; - MqttPendingPublish(int messageId, Promise future, ByteBuf payload, MqttPublishMessage message, MqttQoS qos, PendingOperation operation) { + private MqttPendingPublish( + int messageId, + Promise future, + ByteBuf payload, + MqttPublishMessage message, + MqttQoS qos, + String ownerId, + MqttClientConfig.RetransmissionConfig retransmissionConfig, + PendingOperation pendingOperation + ) { this.messageId = messageId; this.future = future; this.payload = payload; this.message = message; this.qos = qos; - this.publishRetransmissionHandler = new RetransmissionHandler<>(operation); - this.publishRetransmissionHandler.setOriginalMessage(message); - this.pubrelRetransmissionHandler = new RetransmissionHandler<>(operation); - } - - int getMessageId() { - return messageId; - } - - Promise getFuture() { - return future; - } - - ByteBuf getPayload() { - return payload; - } - - boolean isSent() { - return sent; - } - - void setSent(boolean sent) { - this.sent = sent; - } - - MqttPublishMessage getMessage() { - return message; - } - - MqttQoS getQos() { - return qos; + publishRetransmissionHandler = new RetransmissionHandler<>(retransmissionConfig, pendingOperation, ownerId); + publishRetransmissionHandler.setOriginalMessage(message); + pubrelRetransmissionHandler = new RetransmissionHandler<>(retransmissionConfig, pendingOperation, ownerId); } void startPublishRetransmissionTimer(EventLoop eventLoop, Consumer sendPacket) { - this.publishRetransmissionHandler.setHandle(((fixedHeader, originalMessage) -> - sendPacket.accept(new MqttPublishMessage(fixedHeader, originalMessage.variableHeader(), this.payload.retain())))); - this.publishRetransmissionHandler.start(eventLoop); + publishRetransmissionHandler.setHandler(((fixedHeader, originalMessage) -> + sendPacket.accept(new MqttPublishMessage(fixedHeader, originalMessage.variableHeader(), payload.retain())))); + publishRetransmissionHandler.start(eventLoop); } void onPubackReceived() { - this.publishRetransmissionHandler.stop(); + publishRetransmissionHandler.stop(); } void setPubrelMessage(MqttMessage pubrelMessage) { - this.pubrelRetransmissionHandler.setOriginalMessage(pubrelMessage); + pubrelRetransmissionHandler.setOriginalMessage(pubrelMessage); } void startPubrelRetransmissionTimer(EventLoop eventLoop, Consumer sendPacket) { - this.pubrelRetransmissionHandler.setHandle((fixedHeader, originalMessage) -> + pubrelRetransmissionHandler.setHandler((fixedHeader, originalMessage) -> sendPacket.accept(new MqttMessage(fixedHeader, originalMessage.variableHeader()))); - this.pubrelRetransmissionHandler.start(eventLoop); + pubrelRetransmissionHandler.start(eventLoop); } void onPubcompReceived() { - this.pubrelRetransmissionHandler.stop(); + pubrelRetransmissionHandler.stop(); } void onChannelClosed() { - this.publishRetransmissionHandler.stop(); - this.pubrelRetransmissionHandler.stop(); + publishRetransmissionHandler.stop(); + pubrelRetransmissionHandler.stop(); if (payload != null) { payload.release(); } } + + static Builder builder() { + return new Builder(); + } + + static class Builder { + + private int messageId; + private Promise future; + private ByteBuf payload; + private MqttPublishMessage message; + private MqttQoS qos; + private String ownerId; + private MqttClientConfig.RetransmissionConfig retransmissionConfig; + private PendingOperation pendingOperation; + + Builder messageId(int messageId) { + this.messageId = messageId; + return this; + } + + Builder future(Promise future) { + this.future = future; + return this; + } + + Builder payload(ByteBuf payload) { + this.payload = payload; + return this; + } + + Builder message(MqttPublishMessage message) { + this.message = message; + return this; + } + + Builder qos(MqttQoS qos) { + this.qos = qos; + return this; + } + + Builder ownerId(String ownerId) { + this.ownerId = ownerId; + return this; + } + + Builder retransmissionConfig(MqttClientConfig.RetransmissionConfig retransmissionConfig) { + this.retransmissionConfig = retransmissionConfig; + return this; + } + + Builder pendingOperation(PendingOperation pendingOperation) { + this.pendingOperation = pendingOperation; + return this; + } + + MqttPendingPublish build() { + return new MqttPendingPublish(messageId, future, payload, message, qos, ownerId, retransmissionConfig, pendingOperation); + } + + } + } diff --git a/netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttPendingSubscription.java b/netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttPendingSubscription.java index af5d53a06c..7b2ba613cb 100644 --- a/netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttPendingSubscription.java +++ b/netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttPendingSubscription.java @@ -18,90 +18,123 @@ package org.thingsboard.mqtt; import io.netty.channel.EventLoop; import io.netty.handler.codec.mqtt.MqttSubscribeMessage; import io.netty.util.concurrent.Promise; +import lombok.AccessLevel; +import lombok.Getter; +import lombok.Setter; import java.util.HashSet; import java.util.Set; import java.util.function.Consumer; +import static java.util.Objects.requireNonNullElseGet; + +@Getter(AccessLevel.PACKAGE) final class MqttPendingSubscription { private final Promise future; private final String topic; - private final Set handlers = new HashSet<>(); + private final Set handlers; private final MqttSubscribeMessage subscribeMessage; + @Getter(AccessLevel.NONE) private final RetransmissionHandler retransmissionHandler; + @Setter(AccessLevel.PACKAGE) private boolean sent = false; - MqttPendingSubscription(Promise future, String topic, MqttSubscribeMessage message, PendingOperation operation) { + private MqttPendingSubscription( + Promise future, + String topic, + Set handlers, + MqttSubscribeMessage subscribeMessage, + String ownerId, + MqttClientConfig.RetransmissionConfig retransmissionConfig, + PendingOperation operation + ) { this.future = future; this.topic = topic; - this.subscribeMessage = message; + this.handlers = requireNonNullElseGet(handlers, HashSet::new); + this.subscribeMessage = subscribeMessage; - this.retransmissionHandler = new RetransmissionHandler<>(operation); - this.retransmissionHandler.setOriginalMessage(message); + retransmissionHandler = new RetransmissionHandler<>(retransmissionConfig, operation, ownerId); + retransmissionHandler.setOriginalMessage(subscribeMessage); } - Promise getFuture() { - return future; - } + record MqttPendingHandler(MqttHandler handler, boolean once) {} - String getTopic() { - return topic; + void addHandler(MqttHandler handler, boolean once) { + handlers.add(new MqttPendingHandler(handler, once)); } - boolean isSent() { - return sent; + void startRetransmitTimer(EventLoop eventLoop, Consumer sendPacket) { + if (sent) { // If the packet is sent, we can start the retransmission timer + retransmissionHandler.setHandler((fixedHeader, originalMessage) -> + sendPacket.accept(new MqttSubscribeMessage(fixedHeader, originalMessage.variableHeader(), originalMessage.payload()))); + retransmissionHandler.start(eventLoop); + } } - void setSent(boolean sent) { - this.sent = sent; + void onSubackReceived() { + retransmissionHandler.stop(); } - MqttSubscribeMessage getSubscribeMessage() { - return subscribeMessage; + void onChannelClosed() { + retransmissionHandler.stop(); } - void addHandler(MqttHandler handler, boolean once) { - this.handlers.add(new MqttPendingHandler(handler, once)); + static Builder builder() { + return new Builder(); } - Set getHandlers() { - return handlers; - } + static class Builder { - void startRetransmitTimer(EventLoop eventLoop, Consumer sendPacket) { - if (this.sent) { //If the packet is sent, we can start the retransmit timer - this.retransmissionHandler.setHandle((fixedHeader, originalMessage) -> - sendPacket.accept(new MqttSubscribeMessage(fixedHeader, originalMessage.variableHeader(), originalMessage.payload()))); - this.retransmissionHandler.start(eventLoop); + private Promise future; + private String topic; + private Set handlers; + private MqttSubscribeMessage subscribeMessage; + private String ownerId; + private PendingOperation pendingOperation; + private MqttClientConfig.RetransmissionConfig retransmissionConfig; + + Builder future(Promise future) { + this.future = future; + return this; } - } - void onSubackReceived() { - this.retransmissionHandler.stop(); - } + Builder topic(String topic) { + this.topic = topic; + return this; + } - final class MqttPendingHandler { - private final MqttHandler handler; - private final boolean once; + Builder handlers(Set handlers) { + this.handlers = handlers; + return this; + } - MqttPendingHandler(MqttHandler handler, boolean once) { - this.handler = handler; - this.once = once; + Builder subscribeMessage(MqttSubscribeMessage subscribeMessage) { + this.subscribeMessage = subscribeMessage; + return this; } - MqttHandler getHandler() { - return handler; + Builder ownerId(String ownerId) { + this.ownerId = ownerId; + return this; } - boolean isOnce() { - return once; + Builder retransmissionConfig(MqttClientConfig.RetransmissionConfig retransmissionConfig) { + this.retransmissionConfig = retransmissionConfig; + return this; + } + + Builder pendingOperation(PendingOperation pendingOperation) { + this.pendingOperation = pendingOperation; + return this; + } + + MqttPendingSubscription build() { + return new MqttPendingSubscription(future, topic, handlers, subscribeMessage, ownerId, retransmissionConfig, pendingOperation); } - } - void onChannelClosed() { - this.retransmissionHandler.stop(); } + } diff --git a/netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttPendingUnsubscription.java b/netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttPendingUnsubscription.java index 9cb3bd2f8d..8bc23292f8 100644 --- a/netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttPendingUnsubscription.java +++ b/netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttPendingUnsubscription.java @@ -18,43 +18,96 @@ package org.thingsboard.mqtt; import io.netty.channel.EventLoop; import io.netty.handler.codec.mqtt.MqttUnsubscribeMessage; import io.netty.util.concurrent.Promise; +import lombok.AccessLevel; +import lombok.Getter; import java.util.function.Consumer; -final class MqttPendingUnsubscription{ +@Getter(AccessLevel.PACKAGE) +final class MqttPendingUnsubscription { private final Promise future; private final String topic; + @Getter(AccessLevel.NONE) private final RetransmissionHandler retransmissionHandler; - MqttPendingUnsubscription(Promise future, String topic, MqttUnsubscribeMessage unsubscribeMessage, PendingOperation operation) { + private MqttPendingUnsubscription( + Promise future, + String topic, + MqttUnsubscribeMessage unsubscribeMessage, + String ownerId, + MqttClientConfig.RetransmissionConfig retransmissionConfig, + PendingOperation operation + ) { this.future = future; this.topic = topic; - this.retransmissionHandler = new RetransmissionHandler<>(operation); - this.retransmissionHandler.setOriginalMessage(unsubscribeMessage); + retransmissionHandler = new RetransmissionHandler<>(retransmissionConfig, operation, ownerId); + retransmissionHandler.setOriginalMessage(unsubscribeMessage); } - Promise getFuture() { - return future; + void startRetransmissionTimer(EventLoop eventLoop, Consumer sendPacket) { + retransmissionHandler.setHandler((fixedHeader, originalMessage) -> + sendPacket.accept(new MqttUnsubscribeMessage(fixedHeader, originalMessage.variableHeader(), originalMessage.payload()))); + retransmissionHandler.start(eventLoop); } - String getTopic() { - return topic; + void onUnsubackReceived() { + retransmissionHandler.stop(); } - void startRetransmissionTimer(EventLoop eventLoop, Consumer sendPacket) { - this.retransmissionHandler.setHandle((fixedHeader, originalMessage) -> - sendPacket.accept(new MqttUnsubscribeMessage(fixedHeader, originalMessage.variableHeader(), originalMessage.payload()))); - this.retransmissionHandler.start(eventLoop); + void onChannelClosed() { + retransmissionHandler.stop(); } - void onUnsubackReceived(){ - this.retransmissionHandler.stop(); + static Builder builder() { + return new Builder(); } - void onChannelClosed(){ - this.retransmissionHandler.stop(); + static class Builder { + + private Promise future; + private String topic; + private MqttUnsubscribeMessage unsubscribeMessage; + private String ownerId; + private PendingOperation pendingOperation; + private MqttClientConfig.RetransmissionConfig retransmissionConfig; + + Builder future(Promise future) { + this.future = future; + return this; + } + + Builder topic(String topic) { + this.topic = topic; + return this; + } + + Builder unsubscribeMessage(MqttUnsubscribeMessage unsubscribeMessage) { + this.unsubscribeMessage = unsubscribeMessage; + return this; + } + + Builder ownerId(String ownerId) { + this.ownerId = ownerId; + return this; + } + + Builder retransmissionConfig(MqttClientConfig.RetransmissionConfig retransmissionConfig) { + this.retransmissionConfig = retransmissionConfig; + return this; + } + + Builder pendingOperation(PendingOperation pendingOperation) { + this.pendingOperation = pendingOperation; + return this; + } + + MqttPendingUnsubscription build() { + return new MqttPendingUnsubscription(future, topic, unsubscribeMessage, ownerId, retransmissionConfig, pendingOperation); + } + } + } diff --git a/netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttPingHandler.java b/netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttPingHandler.java index 70a4992d72..3fc2c6246e 100644 --- a/netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttPingHandler.java +++ b/netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttPingHandler.java @@ -42,12 +42,11 @@ final class MqttPingHandler extends ChannelInboundHandlerAdapter { } @Override - public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { - if (!(msg instanceof MqttMessage)) { + public void channelRead(ChannelHandlerContext ctx, Object msg) { + if (!(msg instanceof MqttMessage message)) { ctx.fireChannelRead(msg); return; } - MqttMessage message = (MqttMessage) msg; if (message.fixedHeader().messageType() == MqttMessageType.PINGREQ) { this.handlePingReq(ctx.channel()); } else if (message.fixedHeader().messageType() == MqttMessageType.PINGRESP) { @@ -61,28 +60,29 @@ final class MqttPingHandler extends ChannelInboundHandlerAdapter { public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { super.userEventTriggered(ctx, evt); - if (evt instanceof IdleStateEvent) { - IdleStateEvent event = (IdleStateEvent) evt; + if (evt instanceof IdleStateEvent event) { switch (event.state()) { case READER_IDLE: log.debug("[{}] No reads were performed for specified period for channel {}", event.state(), ctx.channel().id()); - this.sendPingReq(ctx.channel()); + this.sendPingReq(ctx.channel(), event); break; case WRITER_IDLE: log.debug("[{}] No writes were performed for specified period for channel {}", event.state(), ctx.channel().id()); - this.sendPingReq(ctx.channel()); + this.sendPingReq(ctx.channel(), event); break; } } } - private void sendPingReq(Channel channel) { + private void sendPingReq(Channel channel, IdleStateEvent idleEvent) { log.trace("[{}] Sending ping request", channel.id()); MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PINGREQ, false, MqttQoS.AT_MOST_ONCE, false, 0); channel.writeAndFlush(new MqttMessage(fixedHeader)); if (this.pingRespTimeout == null) { + log.trace("[{}] Scheduling disconnect due to {}", channel.id(), idleEvent); this.pingRespTimeout = channel.eventLoop().schedule(() -> { + log.trace("[{}] Sending disconnect due to {}", channel.id(), idleEvent); MqttFixedHeader fixedHeader2 = new MqttFixedHeader(MqttMessageType.DISCONNECT, false, MqttQoS.AT_MOST_ONCE, false, 0); channel.writeAndFlush(new MqttMessage(fixedHeader2)).addListener(ChannelFutureListener.CLOSE); //TODO: what do when the connection is closed ? @@ -99,6 +99,7 @@ final class MqttPingHandler extends ChannelInboundHandlerAdapter { private void handlePingResp(Channel channel) { log.trace("[{}] Handling ping response", channel.id()); if (this.pingRespTimeout != null && !this.pingRespTimeout.isCancelled() && !this.pingRespTimeout.isDone()) { + log.trace("[{}] Cancelling disconnect due to idle event because ping response was received", channel.id()); this.pingRespTimeout.cancel(true); this.pingRespTimeout = null; } diff --git a/netty-mqtt/src/main/java/org/thingsboard/mqtt/PendingOperation.java b/netty-mqtt/src/main/java/org/thingsboard/mqtt/PendingOperation.java index b859b216e6..07e472abb3 100644 --- a/netty-mqtt/src/main/java/org/thingsboard/mqtt/PendingOperation.java +++ b/netty-mqtt/src/main/java/org/thingsboard/mqtt/PendingOperation.java @@ -17,6 +17,8 @@ package org.thingsboard.mqtt; public interface PendingOperation { - boolean isCanceled(); + boolean isCancelled(); + + void onMaxRetransmissionAttemptsReached(); } diff --git a/netty-mqtt/src/main/java/org/thingsboard/mqtt/RetransmissionHandler.java b/netty-mqtt/src/main/java/org/thingsboard/mqtt/RetransmissionHandler.java index b0d9ba9002..1778abc593 100644 --- a/netty-mqtt/src/main/java/org/thingsboard/mqtt/RetransmissionHandler.java +++ b/netty-mqtt/src/main/java/org/thingsboard/mqtt/RetransmissionHandler.java @@ -18,66 +18,119 @@ package org.thingsboard.mqtt; import io.netty.channel.EventLoop; import io.netty.handler.codec.mqtt.MqttFixedHeader; import io.netty.handler.codec.mqtt.MqttMessage; +import io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader; import io.netty.handler.codec.mqtt.MqttMessageType; +import io.netty.handler.codec.mqtt.MqttPublishVariableHeader; import io.netty.handler.codec.mqtt.MqttQoS; import io.netty.util.concurrent.ScheduledFuture; import lombok.RequiredArgsConstructor; +import lombok.Setter; +import lombok.extern.slf4j.Slf4j; +import java.util.concurrent.ThreadLocalRandom; import java.util.concurrent.TimeUnit; import java.util.function.BiConsumer; +@Slf4j @RequiredArgsConstructor final class RetransmissionHandler { - private volatile boolean stopped; + private final MqttClientConfig.RetransmissionConfig config; private final PendingOperation pendingOperation; + + private volatile boolean stopped; private ScheduledFuture timer; - private int timeout = 10; + private int attemptCount = 0; + + @Setter private BiConsumer handler; + + // the three fields below are used for logging only + private final String ownerId; + private String originalMessageId; + private long totalWaitingTimeMillis; + private T originalMessage; + void setOriginalMessage(T originalMessage) { + this.originalMessage = originalMessage; + var variableHeader = originalMessage.variableHeader(); + if (variableHeader instanceof MqttMessageIdVariableHeader messageIdVariableHeader) { + originalMessageId = String.valueOf(messageIdVariableHeader.messageId()); + } else if (variableHeader instanceof MqttPublishVariableHeader publishVariableHeader) { + originalMessageId = String.valueOf(publishVariableHeader.packetId()); + } else { + originalMessageId = "N/A"; + } + } + void start(EventLoop eventLoop) { if (eventLoop == null) { throw new NullPointerException("eventLoop"); } - if (this.handler == null) { + if (handler == null) { throw new NullPointerException("handler"); } - this.timeout = 10; - this.startTimer(eventLoop); + log.debug("{}MessageID[{}] Starting retransmission handler", ownerId, originalMessageId); + startTimer(eventLoop); } private void startTimer(EventLoop eventLoop) { - if (stopped || pendingOperation.isCanceled()) { + if (stopped || pendingOperation.isCancelled()) { return; } - this.timer = eventLoop.schedule(() -> { - if (stopped || pendingOperation.isCanceled()) { + + // Calculate the base delay using exponential backoff. + // For attemptCount == 0, delay = initial delay; for each subsequent attempt, the base delay doubles. + long baseDelay = config.initialDelayMillis() * (long) Math.pow(2, attemptCount); + // Apply jitter: random factor between (1 - jitterFactor) and (1 + jitterFactor). + double minFactor = 1.0 - config.jitterFactor(); + double maxFactor = 1.0 + config.jitterFactor(); + double randomFactor = config.jitterFactor() == 0 ? 1 : ThreadLocalRandom.current().nextDouble(minFactor, maxFactor); + long delayMillisWithJitter = (long) (baseDelay * randomFactor); + totalWaitingTimeMillis += delayMillisWithJitter; + + timer = eventLoop.schedule(() -> { + if (stopped || pendingOperation.isCancelled()) { return; } - this.timeout += 5; - boolean isDup = this.originalMessage.fixedHeader().isDup(); - if (this.originalMessage.fixedHeader().messageType() == MqttMessageType.PUBLISH && this.originalMessage.fixedHeader().qosLevel() != MqttQoS.AT_MOST_ONCE) { - isDup = true; + + attemptCount++; + if (attemptCount > config.maxAttempts()) { + log.debug( + "{}MessageID[{}] Gave up after {} retransmission attempts; waited a total of {} ms without receiving acknowledgement", + ownerId, originalMessageId, config.maxAttempts(), totalWaitingTimeMillis + ); + stop(); + pendingOperation.onMaxRetransmissionAttemptsReached(); + return; } - MqttFixedHeader fixedHeader = new MqttFixedHeader(this.originalMessage.fixedHeader().messageType(), isDup, this.originalMessage.fixedHeader().qosLevel(), this.originalMessage.fixedHeader().isRetain(), this.originalMessage.fixedHeader().remainingLength()); - handler.accept(fixedHeader, originalMessage); + + log.debug("{}MessageID[{}] Retransmission attempt #{} out of {}", ownerId, originalMessageId, attemptCount, config.maxAttempts()); + + var originalFixedHeader = originalMessage.fixedHeader(); + var newFixedHeader = new MqttFixedHeader( + originalFixedHeader.messageType(), + isDup(originalFixedHeader), + originalFixedHeader.qosLevel(), + originalFixedHeader.isRetain(), + originalFixedHeader.remainingLength() + ); + handler.accept(newFixedHeader, originalMessage); startTimer(eventLoop); - }, timeout, TimeUnit.SECONDS); + }, delayMillisWithJitter, TimeUnit.MILLISECONDS); + } + + private static boolean isDup(MqttFixedHeader originalFixedHeader) { + return originalFixedHeader.isDup() || (originalFixedHeader.messageType() == MqttMessageType.PUBLISH && originalFixedHeader.qosLevel() != MqttQoS.AT_MOST_ONCE); } void stop() { + log.debug("{}MessageID[{}] Stopping retransmission handler", ownerId, originalMessageId); stopped = true; - if (this.timer != null) { - this.timer.cancel(true); + if (timer != null) { + timer.cancel(true); } } - void setHandle(BiConsumer runnable) { - this.handler = runnable; - } - - void setOriginalMessage(T originalMessage) { - this.originalMessage = originalMessage; - } } diff --git a/netty-mqtt/src/test/java/org/thingsboard/mqtt/MqttClientTest.java b/netty-mqtt/src/test/java/org/thingsboard/mqtt/MqttClientTest.java new file mode 100644 index 0000000000..1481b354ee --- /dev/null +++ b/netty-mqtt/src/test/java/org/thingsboard/mqtt/MqttClientTest.java @@ -0,0 +1,210 @@ +/** + * Copyright © 2016-2025 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.mqtt; + +import com.google.common.util.concurrent.Futures; +import io.netty.buffer.ByteBuf; +import io.netty.buffer.PooledByteBufAllocator; +import io.netty.handler.codec.mqtt.MqttConnectReturnCode; +import io.netty.handler.codec.mqtt.MqttMessageType; +import io.netty.handler.codec.mqtt.MqttQoS; +import io.netty.util.ResourceLeakDetector; +import io.netty.util.concurrent.Future; +import io.netty.util.concurrent.Promise; +import lombok.extern.slf4j.Slf4j; +import org.awaitility.Awaitility; +import org.awaitility.core.ConditionTimeoutException; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.testcontainers.hivemq.HiveMQContainer; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.junit.jupiter.Testcontainers; +import org.testcontainers.utility.DockerImageName; +import org.thingsboard.common.util.AbstractListeningExecutor; + +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; + +@Slf4j +@Testcontainers +class MqttClientTest { + + final int randomPort = 0; + + @Container + HiveMQContainer broker = new HiveMQContainer(DockerImageName.parse("hivemq/hivemq-ce").withTag("2025.2")); + + MqttTestProxy proxy; + + MqttClient client; + + AbstractListeningExecutor handlerExecutor; + + @BeforeAll + static void init() { + ResourceLeakDetector.setLevel(ResourceLeakDetector.Level.PARANOID); + } + + @BeforeEach + void setup() { + handlerExecutor = new AbstractListeningExecutor() { + @Override + protected int getThreadPollSize() { + return 1; + } + }; + handlerExecutor.init(); + } + + @AfterEach + void cleanup() { + if (client != null) { + client.disconnect(); + client = null; + } + if (proxy != null) { + proxy.stop(); + proxy = null; + } + handlerExecutor.destroy(); + handlerExecutor = null; + } + + @Test + void testConnectToBroker() { + // GIVEN + var clientConfig = new MqttClientConfig(); + clientConfig.setOwnerId("Test[ConnectToBroker]"); + clientConfig.setClientId("connect"); + + client = MqttClient.create(clientConfig, null, handlerExecutor); + + // WHEN + Promise connectFuture = client.connect(broker.getHost(), broker.getMqttPort()); + + // THEN + assertThat(connectFuture).isNotNull(); + + Awaitility.await("waiting for client to connect") + .atMost(Duration.ofSeconds(10L)) + .until(connectFuture::isDone); + + assertThat(connectFuture.isSuccess()).isTrue(); + + MqttConnectResult actualConnectResult = connectFuture.getNow(); + assertThat(actualConnectResult).isNotNull(); + assertThat(actualConnectResult.isSuccess()).isTrue(); + assertThat(actualConnectResult.getReturnCode()).isEqualTo(MqttConnectReturnCode.CONNECTION_ACCEPTED); + + assertThat(client.isConnected()).isTrue(); + } + + @Test + void testDisconnectDueToKeepAliveIfNoActivity() { + // GIVEN + proxy = MqttTestProxy.builder() + .localPort(randomPort) + .brokerHost(broker.getHost()) + .brokerPort(broker.getMqttPort()) + .brokerToClientInterceptor(msg -> msg.fixedHeader().messageType() != MqttMessageType.PINGRESP) // drop all ping responses to simulate broker down + .build(); + + int idleTimeoutSeconds = 2; + + var clientConfig = new MqttClientConfig(); + clientConfig.setOwnerId("Test[KeepAliveDisconnect]"); + clientConfig.setClientId("no-activity-disconnect"); + clientConfig.setTimeoutSeconds(idleTimeoutSeconds); + clientConfig.setReconnect(false); // disable auto reconnect + client = MqttClient.create(clientConfig, null, handlerExecutor); + + // WHEN-THEN + connect(broker.getHost(), proxy.getPort()); + + // no activity... + + Awaitility.await("waiting for client to disconnect") + .pollDelay(Duration.ofSeconds(idleTimeoutSeconds * 2)) // 2 seconds to wait for the first idle event and then 2 seconds for scheduled disconnect to fire + .atMost(Duration.ofSeconds(10)) + .untilAsserted(() -> assertThat(client.isConnected()).isFalse()); + } + + @Test + void testRetransmission() { + // GIVEN + proxy = MqttTestProxy.builder() + .localPort(randomPort) + .brokerHost(broker.getHost()) + .brokerPort(broker.getMqttPort()) + .brokerToClientInterceptor(msg -> msg.fixedHeader().messageType() != MqttMessageType.PUBACK) // drop all pubacks to allow retransmission to happen + .build(); + + // create client + var clientConfig = new MqttClientConfig(); + clientConfig.setOwnerId("Test[Retransmission]"); + clientConfig.setClientId("retransmission"); + clientConfig.setRetransmissionConfig(new MqttClientConfig.RetransmissionConfig(1, 1000L, 0d)); + client = MqttClient.create(clientConfig, null, handlerExecutor); + + // connect to a broker + connect(broker.getHost(), proxy.getPort()); + + // subscribe to a topic + String topic = "test-topic"; + List receivedMessages = Collections.synchronizedList(new ArrayList<>(2)); + Future subscribeFuture = client.on(topic, (__, payload) -> { + receivedMessages.add(payload); + return Futures.immediateVoidFuture(); + }); + Awaitility.await("waiting for client to subscribe to a topic") + .atMost(Duration.ofSeconds(10L)) + .until(subscribeFuture::isDone); + + // WHEN + // publish a message + ByteBuf message = PooledByteBufAllocator.DEFAULT.buffer().writeBytes("test message".getBytes(StandardCharsets.UTF_8)); + client.publish(topic, message, MqttQoS.AT_LEAST_ONCE); + + // THEN + // wait enough time so that retransmission happens and stops + // if retransmission works incorrectly waiting 10 seconds allows for additional retransmissions to happen + try { + Awaitility.await("wait up to 10s, stop early if too many messages") + .atMost(Duration.ofSeconds(10L)) + .pollInterval(Duration.ofMillis(100)) + .until(() -> receivedMessages.size() > 2); + } catch (ConditionTimeoutException __) { + // didn't exceed 2 messages + } + + assertThat(receivedMessages).size().describedAs("incorrect number of messages received, expected 2 (original plus one retransmitted)").isEqualTo(2); + } + + private void connect(String host, int port) { + Promise connectFuture = client.connect(host, port); + Awaitility.await("waiting for client to connect") + .atMost(Duration.ofSeconds(10L)) + .until(connectFuture::isSuccess); + } + +} diff --git a/netty-mqtt/src/test/java/org/thingsboard/mqtt/MqttPingHandlerTest.java b/netty-mqtt/src/test/java/org/thingsboard/mqtt/MqttPingHandlerTest.java deleted file mode 100644 index 83e3b1c8d5..0000000000 --- a/netty-mqtt/src/test/java/org/thingsboard/mqtt/MqttPingHandlerTest.java +++ /dev/null @@ -1,63 +0,0 @@ -/** - * Copyright © 2016-2025 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.mqtt; - -import io.netty.channel.Channel; -import io.netty.channel.ChannelFuture; -import io.netty.channel.ChannelFutureListener; -import io.netty.channel.ChannelHandlerContext; -import io.netty.channel.DefaultEventLoop; -import io.netty.handler.timeout.IdleStateEvent; -import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.api.Test; - -import java.util.concurrent.TimeUnit; - -import static org.mockito.ArgumentMatchers.any; -import static org.mockito.ArgumentMatchers.eq; -import static org.mockito.Mockito.after; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.when; - -class MqttPingHandlerTest { - - static final int KEEP_ALIVE_SECONDS = 0; - static final int PROCESS_SEND_DISCONNECT_MSG_TIME_MS = 500; - - MqttPingHandler mqttPingHandler; - - @BeforeEach - void setUp() { - mqttPingHandler = new MqttPingHandler(KEEP_ALIVE_SECONDS); - } - - @Test - void givenChannelReaderIdleState_whenNoPingResponse_thenDisconnectClient() throws Exception { - ChannelHandlerContext ctx = mock(ChannelHandlerContext.class); - Channel channel = mock(Channel.class); - when(ctx.channel()).thenReturn(channel); - when(channel.eventLoop()).thenReturn(new DefaultEventLoop()); - ChannelFuture channelFuture = mock(ChannelFuture.class); - when(channel.writeAndFlush(any())).thenReturn(channelFuture); - - mqttPingHandler.userEventTriggered(ctx, IdleStateEvent.FIRST_READER_IDLE_STATE_EVENT); - verify( - channelFuture, - after(TimeUnit.SECONDS.toMillis(KEEP_ALIVE_SECONDS) + PROCESS_SEND_DISCONNECT_MSG_TIME_MS) - ).addListener(eq(ChannelFutureListener.CLOSE)); - } -} \ No newline at end of file diff --git a/netty-mqtt/src/test/java/org/thingsboard/mqtt/MqttTestProxy.java b/netty-mqtt/src/test/java/org/thingsboard/mqtt/MqttTestProxy.java new file mode 100644 index 0000000000..4a10fc3bfb --- /dev/null +++ b/netty-mqtt/src/test/java/org/thingsboard/mqtt/MqttTestProxy.java @@ -0,0 +1,202 @@ +/** + * Copyright © 2016-2025 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.mqtt; + +import io.netty.bootstrap.Bootstrap; +import io.netty.bootstrap.ServerBootstrap; +import io.netty.channel.Channel; +import io.netty.channel.ChannelFuture; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelInitializer; +import io.netty.channel.EventLoopGroup; +import io.netty.channel.SimpleChannelInboundHandler; +import io.netty.channel.nio.NioEventLoopGroup; +import io.netty.channel.socket.SocketChannel; +import io.netty.channel.socket.nio.NioServerSocketChannel; +import io.netty.channel.socket.nio.NioSocketChannel; +import io.netty.handler.codec.mqtt.MqttDecoder; +import io.netty.handler.codec.mqtt.MqttEncoder; +import io.netty.handler.codec.mqtt.MqttMessage; +import io.netty.util.ReferenceCountUtil; +import lombok.extern.slf4j.Slf4j; + +import java.net.InetSocketAddress; +import java.util.function.Predicate; + +@Slf4j +public class MqttTestProxy { + + private final EventLoopGroup bossGroup; + private final EventLoopGroup workerGroup; + + private Channel clientToProxyChannel; + private Channel proxyToBrokerChannel; + + private final int assignedPort; + + private boolean stopped; + + private final Predicate brokerToClientInterceptor; + + private MqttTestProxy(Builder builder) { + log.info("Starting MQTT proxy..."); + + brokerToClientInterceptor = builder.brokerToClientInterceptor != null ? builder.brokerToClientInterceptor : msg -> true; + bossGroup = new NioEventLoopGroup(1); + workerGroup = new NioEventLoopGroup(1); + + ServerBootstrap proxyBootstrap = new ServerBootstrap(); + proxyBootstrap.group(bossGroup, workerGroup) + .channel(NioServerSocketChannel.class) + .childHandler(new ChannelInitializer() { + @Override + protected void initChannel(SocketChannel channel) { + clientToProxyChannel = channel; + clientToProxyChannel.config().setAutoRead(false); // do not accept data before we connected to a broker + + connectToBroker(builder.brokerHost, builder.brokerPort).addListener(future -> { + if (future.isSuccess()) { + clientToProxyChannel.pipeline().addLast("mqttDecoder", new MqttDecoder()); + clientToProxyChannel.pipeline().addLast("mqttToBroker", new MqttRelayHandler(proxyToBrokerChannel, null)); + clientToProxyChannel.pipeline().addLast("mqttEncoder", MqttEncoder.INSTANCE); + + clientToProxyChannel.config().setAutoRead(true); // start accepting data for a client + } else { + log.error("Failed to connect to broker", future.cause()); + clientToProxyChannel.close(); + } + }); + } + }); + + try { + Channel proxyChannel = proxyBootstrap.bind(builder.localPort).sync().channel(); + assignedPort = ((InetSocketAddress) proxyChannel.localAddress()).getPort(); + } catch (Exception e) { + log.error("Failed to start MQTT proxy", e); + throw new RuntimeException("Failed to start MQTT proxy", e); + } + + log.info("MQTT proxy started on port {}", assignedPort); + } + + private ChannelFuture connectToBroker(String brokerHost, int brokerPort) { + Bootstrap proxyToBrokerBootstrap = new Bootstrap(); + proxyToBrokerBootstrap.group(workerGroup) + .channel(NioSocketChannel.class) + .handler(new ChannelInitializer() { + @Override + protected void initChannel(SocketChannel channel) { + proxyToBrokerChannel = channel; + proxyToBrokerChannel.pipeline().addLast(new MqttDecoder()); + proxyToBrokerChannel.pipeline().addLast("mqttToClient", new MqttRelayHandler(clientToProxyChannel, brokerToClientInterceptor)); + proxyToBrokerChannel.pipeline().addLast(MqttEncoder.INSTANCE); + } + }); + return proxyToBrokerBootstrap.connect(brokerHost, brokerPort); + } + + private static class MqttRelayHandler extends SimpleChannelInboundHandler { + + private final Channel targetChannel; + private final Predicate interceptor; + + private MqttRelayHandler(Channel targetChannel, Predicate interceptor) { + this.targetChannel = targetChannel; + this.interceptor = interceptor; + } + + @Override + protected void channelRead0(ChannelHandlerContext ctx, MqttMessage msg) { + log.debug("Received message: {}", msg.fixedHeader().messageType()); + if (interceptor == null || interceptor.test(msg)) { + if (targetChannel.isActive()) { + targetChannel.writeAndFlush(ReferenceCountUtil.retain(msg)); + } + } else { + log.info("Dropping message: {}", msg.fixedHeader().messageType()); + } + } + + } + + public void stop() { + if (stopped) { + log.info("MQTT proxy was already stopped"); + return; + } + + stopped = true; + + log.info("Stopping MQTT proxy..."); + + if (clientToProxyChannel != null) { + clientToProxyChannel.close(); + } + if (proxyToBrokerChannel != null) { + proxyToBrokerChannel.close(); + } + if (bossGroup != null) { + bossGroup.shutdownGracefully(); + } + if (workerGroup != null) { + workerGroup.shutdownGracefully(); + } + + log.info("MQTT proxy stopped"); + } + + public int getPort() { + return assignedPort; + } + + public static Builder builder() { + return new Builder(); + } + + public static class Builder { + + private int localPort; + private String brokerHost; + private int brokerPort; + private Predicate brokerToClientInterceptor; + + public Builder localPort(int localPort) { + this.localPort = localPort; + return this; + } + + public Builder brokerHost(String brokerHost) { + this.brokerHost = brokerHost; + return this; + } + + public Builder brokerPort(int brokerPort) { + this.brokerPort = brokerPort; + return this; + } + + public Builder brokerToClientInterceptor(Predicate interceptor) { + this.brokerToClientInterceptor = interceptor; + return this; + } + + public MqttTestProxy build() { + return new MqttTestProxy(this); + } + + } +} diff --git a/netty-mqtt/src/test/java/org/thingsboard/mqtt/integration/MqttIntegrationTest.java b/netty-mqtt/src/test/java/org/thingsboard/mqtt/integration/MqttIntegrationTest.java deleted file mode 100644 index db177c84b6..0000000000 --- a/netty-mqtt/src/test/java/org/thingsboard/mqtt/integration/MqttIntegrationTest.java +++ /dev/null @@ -1,151 +0,0 @@ -/** - * Copyright © 2016-2025 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.mqtt.integration; - -import io.netty.buffer.Unpooled; -import io.netty.channel.EventLoopGroup; -import io.netty.channel.nio.NioEventLoopGroup; -import io.netty.handler.codec.mqtt.MqttMessageType; -import io.netty.handler.codec.mqtt.MqttQoS; -import io.netty.util.concurrent.Future; -import io.netty.util.concurrent.Promise; -import lombok.extern.slf4j.Slf4j; -import org.junit.jupiter.api.AfterEach; -import org.junit.jupiter.api.Assertions; -import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.api.Test; -import org.junit.jupiter.api.parallel.ResourceLock; -import org.thingsboard.common.util.AbstractListeningExecutor; -import org.thingsboard.mqtt.MqttClient; -import org.thingsboard.mqtt.MqttClientConfig; -import org.thingsboard.mqtt.MqttConnectResult; -import org.thingsboard.mqtt.integration.server.MqttServer; - -import java.nio.charset.StandardCharsets; -import java.util.List; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; - -@ResourceLock("port8885") // test MQTT server port -@Slf4j -public class MqttIntegrationTest { - - static final String MQTT_HOST = "localhost"; - static final int KEEPALIVE_TIMEOUT_SECONDS = 2; - static final long RECONNECT_DELAY_SECONDS = 10L; - - EventLoopGroup eventLoopGroup; - MqttServer mqttServer; - - MqttClient mqttClient; - - AbstractListeningExecutor handlerExecutor; - - @BeforeEach - public void init() throws Exception { - this.handlerExecutor = new AbstractListeningExecutor() { - @Override - protected int getThreadPollSize() { - return 4; - } - }; - handlerExecutor.init(); - - this.eventLoopGroup = new NioEventLoopGroup(); - - this.mqttServer = new MqttServer(); - this.mqttServer.init(); - } - - @AfterEach - public void destroy() throws InterruptedException { - if (this.mqttClient != null) { - this.mqttClient.disconnect(); - } - if (this.mqttServer != null) { - this.mqttServer.shutdown(); - } - if (this.eventLoopGroup != null) { - this.eventLoopGroup.shutdownGracefully(0, 0, TimeUnit.MILLISECONDS); - } - if (this.handlerExecutor != null) { - this.handlerExecutor.destroy(); - } - } - - @Test - public void givenActiveMqttClient_whenNoActivityForKeepAliveTimeout_thenDisconnectClient() throws Throwable { - //given - this.mqttClient = initClient(); - - log.warn("Sending publish messages..."); - CountDownLatch latch = new CountDownLatch(3); - for (int i = 0; i < 3; i++) { - Thread.sleep(30); - Future pubFuture = publishMsg(); - pubFuture.addListener(future -> latch.countDown()); - } - - log.warn("Waiting for messages acknowledgments..."); - boolean awaitResult = latch.await(10, TimeUnit.SECONDS); - Assertions.assertTrue(awaitResult); - log.warn("Messages are delivered successfully..."); - - //when - log.warn("Starting idle period..."); - Thread.sleep(5000); - - //then - List allReceivedEvents = this.mqttServer.getEventsFromClient(); - long disconnectCount = allReceivedEvents.stream().filter(type -> type == MqttMessageType.DISCONNECT).count(); - - Assertions.assertEquals(1, disconnectCount); - } - - private Future publishMsg() { - return this.mqttClient.publish( - "test/topic", - Unpooled.wrappedBuffer("payload".getBytes(StandardCharsets.UTF_8)), - MqttQoS.AT_MOST_ONCE); - } - - private MqttClient initClient() throws Exception { - MqttClientConfig config = new MqttClientConfig(); - config.setOwnerId("MqttIntegrationTest"); - config.setTimeoutSeconds(KEEPALIVE_TIMEOUT_SECONDS); - config.setReconnectDelay(RECONNECT_DELAY_SECONDS); - MqttClient client = MqttClient.create(config, null, handlerExecutor); - client.setEventLoop(this.eventLoopGroup); - Promise connectFuture = client.connect(MQTT_HOST, this.mqttServer.getMqttPort()); - - String hostPort = MQTT_HOST + ":" + this.mqttServer.getMqttPort(); - MqttConnectResult result; - try { - result = connectFuture.get(10, TimeUnit.SECONDS); - } catch (TimeoutException ex) { - connectFuture.cancel(true); - client.disconnect(); - throw new RuntimeException(String.format("Failed to connect to MQTT server at %s.", hostPort)); - } - if (!result.isSuccess()) { - connectFuture.cancel(true); - client.disconnect(); - throw new RuntimeException(String.format("Failed to connect to MQTT server at %s. Result code is: %s", hostPort, result.getReturnCode())); - } - return client; - } -} \ No newline at end of file diff --git a/netty-mqtt/src/test/java/org/thingsboard/mqtt/integration/server/MqttServer.java b/netty-mqtt/src/test/java/org/thingsboard/mqtt/integration/server/MqttServer.java deleted file mode 100644 index ca4fb677dc..0000000000 --- a/netty-mqtt/src/test/java/org/thingsboard/mqtt/integration/server/MqttServer.java +++ /dev/null @@ -1,84 +0,0 @@ -/** - * Copyright © 2016-2025 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.mqtt.integration.server; - -import io.netty.bootstrap.ServerBootstrap; -import io.netty.channel.Channel; -import io.netty.channel.ChannelInitializer; -import io.netty.channel.ChannelOption; -import io.netty.channel.ChannelPipeline; -import io.netty.channel.EventLoopGroup; -import io.netty.channel.nio.NioEventLoopGroup; -import io.netty.channel.socket.SocketChannel; -import io.netty.channel.socket.nio.NioServerSocketChannel; -import io.netty.handler.codec.mqtt.MqttDecoder; -import io.netty.handler.codec.mqtt.MqttEncoder; -import io.netty.handler.codec.mqtt.MqttMessageType; -import lombok.Getter; -import lombok.extern.slf4j.Slf4j; - -import java.util.List; -import java.util.concurrent.CopyOnWriteArrayList; - -@Slf4j -public class MqttServer { - - @Getter - private final List eventsFromClient = new CopyOnWriteArrayList<>(); - @Getter - private final int mqttPort = 8885; - - private Channel serverChannel; - private EventLoopGroup bossGroup; - private EventLoopGroup workerGroup; - - public void init() throws Exception { - log.info("Starting MQTT server on port {}...", mqttPort); - bossGroup = new NioEventLoopGroup(); - workerGroup = new NioEventLoopGroup(); - ServerBootstrap b = new ServerBootstrap(); - b.group(bossGroup, workerGroup) - .channel(NioServerSocketChannel.class) - .childHandler(new ChannelInitializer() { - @Override - protected void initChannel(SocketChannel ch) throws Exception { - ChannelPipeline pipeline = ch.pipeline(); - pipeline.addLast("decoder", new MqttDecoder(65536)); - pipeline.addLast("encoder", MqttEncoder.INSTANCE); - - MqttTransportHandler handler = new MqttTransportHandler(eventsFromClient); - - pipeline.addLast(handler); - ch.closeFuture().addListener(handler); - } - }) - .childOption(ChannelOption.SO_KEEPALIVE, true); - - serverChannel = b.bind(mqttPort).sync().channel(); - log.info("Mqtt transport started!"); - } - - public void shutdown() throws InterruptedException { - log.info("Stopping MQTT transport!"); - try { - serverChannel.close().sync(); - } finally { - workerGroup.shutdownGracefully(); - bossGroup.shutdownGracefully(); - } - log.info("MQTT transport stopped!"); - } -} diff --git a/netty-mqtt/src/test/java/org/thingsboard/mqtt/integration/server/MqttTransportHandler.java b/netty-mqtt/src/test/java/org/thingsboard/mqtt/integration/server/MqttTransportHandler.java deleted file mode 100644 index 5c433d7069..0000000000 --- a/netty-mqtt/src/test/java/org/thingsboard/mqtt/integration/server/MqttTransportHandler.java +++ /dev/null @@ -1,141 +0,0 @@ -/** - * Copyright © 2016-2025 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.mqtt.integration.server; - -import io.netty.channel.ChannelHandlerContext; -import io.netty.channel.ChannelInboundHandlerAdapter; -import io.netty.handler.codec.mqtt.MqttConnAckMessage; -import io.netty.handler.codec.mqtt.MqttConnAckVariableHeader; -import io.netty.handler.codec.mqtt.MqttConnectMessage; -import io.netty.handler.codec.mqtt.MqttConnectReturnCode; -import io.netty.handler.codec.mqtt.MqttFixedHeader; -import io.netty.handler.codec.mqtt.MqttMessage; -import io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader; -import io.netty.handler.codec.mqtt.MqttMessageType; -import io.netty.handler.codec.mqtt.MqttPubAckMessage; -import io.netty.handler.codec.mqtt.MqttPublishMessage; -import io.netty.util.ReferenceCountUtil; -import io.netty.util.concurrent.Future; -import io.netty.util.concurrent.GenericFutureListener; -import lombok.extern.slf4j.Slf4j; - -import java.util.List; -import java.util.UUID; - -import static io.netty.handler.codec.mqtt.MqttMessageType.CONNACK; -import static io.netty.handler.codec.mqtt.MqttMessageType.CONNECT; -import static io.netty.handler.codec.mqtt.MqttMessageType.DISCONNECT; -import static io.netty.handler.codec.mqtt.MqttMessageType.PINGREQ; -import static io.netty.handler.codec.mqtt.MqttMessageType.PUBACK; -import static io.netty.handler.codec.mqtt.MqttMessageType.PUBLISH; -import static io.netty.handler.codec.mqtt.MqttQoS.AT_MOST_ONCE; - -@Slf4j -public class MqttTransportHandler extends ChannelInboundHandlerAdapter implements GenericFutureListener> { - - private final List eventsFromClient; - private final UUID sessionId; - - MqttTransportHandler(List eventsFromClient) { - this.sessionId = UUID.randomUUID(); - this.eventsFromClient = eventsFromClient; - } - - @Override - public void channelRead(ChannelHandlerContext ctx, Object msg) { - log.trace("[{}] Processing msg: {}", sessionId, msg); - try { - if (msg instanceof MqttMessage) { - MqttMessage message = (MqttMessage) msg; - if (message.decoderResult().isSuccess()) { - processMqttMsg(ctx, message); - } else { - log.error("[{}] Message decoding failed: {}", sessionId, message.decoderResult().cause().getMessage()); - ctx.close(); - } - } else { - log.debug("[{}] Received non mqtt message: {}", sessionId, msg.getClass().getSimpleName()); - ctx.close(); - } - } finally { - ReferenceCountUtil.safeRelease(msg); - } - } - - void processMqttMsg(ChannelHandlerContext ctx, MqttMessage msg) { - if (msg.fixedHeader() == null) { - ctx.close(); - return; - } - switch (msg.fixedHeader().messageType()) { - case CONNECT: - eventsFromClient.add(CONNECT); - processConnect(ctx, (MqttConnectMessage) msg); - break; - case DISCONNECT: - eventsFromClient.add(DISCONNECT); - ctx.close(); - break; - case PUBLISH: - // QoS 0 and 1 supported only here - eventsFromClient.add(PUBLISH); - MqttPublishMessage mqttPubMsg = (MqttPublishMessage) msg; - ack(ctx, mqttPubMsg.variableHeader().packetId()); - break; - case PINGREQ: - // We will not handle PINGREQ and will not send any PINGRESP to simulate the MQTT server is down - eventsFromClient.add(PINGREQ); - break; - default: - break; - } - } - - void processConnect(ChannelHandlerContext ctx, MqttConnectMessage msg) { - String userName = msg.payload().userName(); - String clientId = msg.payload().clientIdentifier(); - - log.warn("[{}][{}] Processing connect msg for client: {}!", sessionId, userName, clientId); - ctx.writeAndFlush(createMqttConnAckMsg(msg)); - } - - private MqttConnAckMessage createMqttConnAckMsg(MqttConnectMessage msg) { - MqttFixedHeader mqttFixedHeader = - new MqttFixedHeader(CONNACK, false, AT_MOST_ONCE, false, 0); - MqttConnAckVariableHeader mqttConnAckVariableHeader = - new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_ACCEPTED, !msg.variableHeader().isCleanSession()); - return new MqttConnAckMessage(mqttFixedHeader, mqttConnAckVariableHeader); - } - - private void ack(ChannelHandlerContext ctx, int msgId) { - if (msgId > 0) { - ctx.writeAndFlush(createMqttPubAckMsg(msgId)); - } - } - - public static MqttPubAckMessage createMqttPubAckMsg(int requestId) { - MqttFixedHeader mqttFixedHeader = - new MqttFixedHeader(PUBACK, false, AT_MOST_ONCE, false, 0); - MqttMessageIdVariableHeader mqttMsgIdVariableHeader = - MqttMessageIdVariableHeader.from(requestId); - return new MqttPubAckMessage(mqttFixedHeader, mqttMsgIdVariableHeader); - } - - @Override - public void operationComplete(Future future) { - log.trace("[{}] Channel closed!", sessionId); - } -} diff --git a/netty-mqtt/src/test/resources/junit-platform.properties b/netty-mqtt/src/test/resources/junit-platform.properties deleted file mode 100644 index f2ed301920..0000000000 --- a/netty-mqtt/src/test/resources/junit-platform.properties +++ /dev/null @@ -1,3 +0,0 @@ -junit.jupiter.execution.parallel.enabled = true -junit.jupiter.execution.parallel.mode.default = concurrent -junit.jupiter.execution.parallel.mode.classes.default = concurrent diff --git a/pom.xml b/pom.xml index 6d7981ccbb..f95e9df9b1 100755 --- a/pom.xml +++ b/pom.xml @@ -1973,6 +1973,12 @@ ${testcontainers.version} test + + org.testcontainers + hivemq + ${testcontainers.version} + test + org.springframework.data spring-data-redis diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/MqttClientSettings.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/MqttClientSettings.java new file mode 100644 index 0000000000..4ac05b57d0 --- /dev/null +++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/MqttClientSettings.java @@ -0,0 +1,26 @@ +/** + * Copyright © 2016-2025 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.rule.engine.api; + +public interface MqttClientSettings { + + int getRetransmissionMaxAttempts(); + + long getRetransmissionInitialDelayMillis(); + + double getRetransmissionJitterFactor(); + +} 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 b66c9e13d5..7989b8f9ce 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 @@ -416,4 +416,9 @@ public interface TbContext { EventService getEventService(); AuditLogService getAuditLogService(); + + // Configuration parameters for the MQTT client that is used in the MQTT node and Azure IoT hub node + + MqttClientSettings getMqttClientSettings(); + } diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/mqtt/TbMqttNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/mqtt/TbMqttNode.java index 4d99951e1a..28a9e1ff4b 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/mqtt/TbMqttNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/mqtt/TbMqttNode.java @@ -26,6 +26,7 @@ import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.mqtt.MqttClient; import org.thingsboard.mqtt.MqttClientConfig; import org.thingsboard.mqtt.MqttConnectResult; +import org.thingsboard.rule.engine.api.MqttClientSettings; import org.thingsboard.rule.engine.api.RuleNode; import org.thingsboard.rule.engine.api.TbContext; import org.thingsboard.rule.engine.api.TbNodeConfiguration; @@ -126,6 +127,13 @@ public class TbMqttNode extends TbAbstractExternalNode { } config.setCleanSession(this.mqttNodeConfiguration.isCleanSession()); + MqttClientSettings mqttClientSettings = ctx.getMqttClientSettings(); + config.setRetransmissionConfig(new MqttClientConfig.RetransmissionConfig( + mqttClientSettings.getRetransmissionMaxAttempts(), + mqttClientSettings.getRetransmissionInitialDelayMillis(), + mqttClientSettings.getRetransmissionJitterFactor() + )); + prepareMqttClientConfig(config); MqttClient client = getMqttClient(ctx, config); client.setEventLoop(ctx.getSharedEventLoop()); diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/mqtt/TbMqttNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/mqtt/TbMqttNodeTest.java index f6ccfbca6f..bf650af8bb 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/mqtt/TbMqttNodeTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/mqtt/TbMqttNodeTest.java @@ -40,6 +40,7 @@ import org.thingsboard.mqtt.MqttClient; import org.thingsboard.mqtt.MqttClientConfig; import org.thingsboard.mqtt.MqttConnectResult; import org.thingsboard.rule.engine.AbstractRuleNodeUpgradeTest; +import org.thingsboard.rule.engine.api.MqttClientSettings; import org.thingsboard.rule.engine.api.TbContext; import org.thingsboard.rule.engine.api.TbNode; import org.thingsboard.rule.engine.api.TbNodeConfiguration; @@ -80,6 +81,7 @@ import static org.mockito.BDDMockito.spy; import static org.mockito.BDDMockito.then; import static org.mockito.BDDMockito.willAnswer; import static org.mockito.BDDMockito.willReturn; +import static org.mockito.Mockito.lenient; @ExtendWith(MockitoExtension.class) public class TbMqttNodeTest extends AbstractRuleNodeUpgradeTest { @@ -106,6 +108,22 @@ public class TbMqttNodeTest extends AbstractRuleNodeUpgradeTest { protected void setUp() { mqttNode = spy(new TbMqttNode()); mqttNodeConfig = new TbMqttNodeConfiguration().defaultConfiguration(); + lenient().when(ctxMock.getMqttClientSettings()).thenReturn(new MqttClientSettings() { + @Override + public int getRetransmissionMaxAttempts() { + return 3; + } + + @Override + public long getRetransmissionInitialDelayMillis() { + return 5000L; + } + + @Override + public double getRetransmissionJitterFactor() { + return 0.15; + } + }); } @Test