Dmytro Skarzhynets
1 year ago
33 changed files with 1100 additions and 623 deletions
@ -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; |
|||
|
|||
} |
|||
@ -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(); |
|||
} |
|||
|
|||
} |
|||
@ -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); |
|||
} |
|||
|
|||
} |
|||
@ -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<MqttConnectResult> 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<ByteBuf> receivedMessages = Collections.synchronizedList(new ArrayList<>(2)); |
|||
Future<Void> 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<MqttConnectResult> connectFuture = client.connect(host, port); |
|||
Awaitility.await("waiting for client to connect") |
|||
.atMost(Duration.ofSeconds(10L)) |
|||
.until(connectFuture::isSuccess); |
|||
} |
|||
|
|||
} |
|||
@ -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)); |
|||
} |
|||
} |
|||
@ -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<MqttMessage> 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<SocketChannel>() { |
|||
@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<SocketChannel>() { |
|||
@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<MqttMessage> { |
|||
|
|||
private final Channel targetChannel; |
|||
private final Predicate<MqttMessage> interceptor; |
|||
|
|||
private MqttRelayHandler(Channel targetChannel, Predicate<MqttMessage> 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<MqttMessage> 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<MqttMessage> interceptor) { |
|||
this.brokerToClientInterceptor = interceptor; |
|||
return this; |
|||
} |
|||
|
|||
public MqttTestProxy build() { |
|||
return new MqttTestProxy(this); |
|||
} |
|||
|
|||
} |
|||
} |
|||
@ -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<Void> 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<MqttMessageType> allReceivedEvents = this.mqttServer.getEventsFromClient(); |
|||
long disconnectCount = allReceivedEvents.stream().filter(type -> type == MqttMessageType.DISCONNECT).count(); |
|||
|
|||
Assertions.assertEquals(1, disconnectCount); |
|||
} |
|||
|
|||
private Future<Void> 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<MqttConnectResult> 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; |
|||
} |
|||
} |
|||
@ -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<MqttMessageType> 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<SocketChannel>() { |
|||
@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!"); |
|||
} |
|||
} |
|||
@ -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<Future<? super Void>> { |
|||
|
|||
private final List<MqttMessageType> eventsFromClient; |
|||
private final UUID sessionId; |
|||
|
|||
MqttTransportHandler(List<MqttMessageType> 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<? super Void> future) { |
|||
log.trace("[{}] Channel closed!", sessionId); |
|||
} |
|||
} |
|||
@ -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 |
|||
@ -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(); |
|||
|
|||
} |
|||
Loading…
Reference in new issue