diff --git a/netty-mqtt/pom.xml b/netty-mqtt/pom.xml index ade1d67bd5..472ce55fc7 100644 --- a/netty-mqtt/pom.xml +++ b/netty-mqtt/pom.xml @@ -84,6 +84,11 @@ awaitility test + + io.takari.junit + takari-cpsuite + test + diff --git a/netty-mqtt/src/test/java/org/thingsboard/mqtt/integration/IntegrationTestSuite.java b/netty-mqtt/src/test/java/org/thingsboard/mqtt/integration/IntegrationTestSuite.java new file mode 100644 index 0000000000..17392f0763 --- /dev/null +++ b/netty-mqtt/src/test/java/org/thingsboard/mqtt/integration/IntegrationTestSuite.java @@ -0,0 +1,27 @@ +/** + * Copyright © 2016-2022 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 org.junit.extensions.cpsuite.ClasspathSuite; +import org.junit.runner.RunWith; + +@RunWith(ClasspathSuite.class) +@ClasspathSuite.ClassnameFilters({ + "org.thingsboard.mqtt.integration.*Test", +}) +public class IntegrationTestSuite { + +} \ No newline at end of file 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 new file mode 100644 index 0000000000..9b665141f3 --- /dev/null +++ b/netty-mqtt/src/test/java/org/thingsboard/mqtt/integration/MqttIntegrationTest.java @@ -0,0 +1,139 @@ +/** + * Copyright © 2016-2022 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.ByteBuf; +import io.netty.buffer.ByteBufAllocator; +import io.netty.buffer.UnpooledByteBufAllocator; +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 lombok.extern.slf4j.Slf4j; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +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; + +@Slf4j +public class MqttIntegrationTest { + + static final String MQTT_HOST = "localhost"; + static final int KEEPALIVE_TIMEOUT_SECONDS = 2; + static final ByteBufAllocator ALLOCATOR = new UnpooledByteBufAllocator(false); + + EventLoopGroup eventLoopGroup; + MqttServer mqttServer; + + MqttClient mqttClient; + + @Before + public void init() throws Exception { + this.eventLoopGroup = new NioEventLoopGroup(); + + this.mqttServer = new MqttServer(); + this.mqttServer.init(); + } + + @After + 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, 5, TimeUnit.SECONDS); + } + } + + @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++) { + Future pubFuture = publishMsg(); + pubFuture.addListener(future -> latch.countDown()); + } + + log.warn("Waiting for messages acknowledgments..."); + boolean awaitResult = latch.await(10, TimeUnit.SECONDS); + Assert.assertTrue(awaitResult); + + //when + CountDownLatch keepAliveLatch = new CountDownLatch(1); + + log.warn("Starting idle period..."); + boolean keepaliveAwaitResult = keepAliveLatch.await(5, TimeUnit.SECONDS); + Assert.assertFalse(keepaliveAwaitResult); + + //then + List allReceivedEvents = this.mqttServer.getEventsFromClient(); + long pubCount = allReceivedEvents.stream().filter(mqttMessageType -> mqttMessageType == MqttMessageType.PUBLISH).count(); + long disconnectCount = allReceivedEvents.stream().filter(type -> type == MqttMessageType.DISCONNECT).count(); + + Assert.assertEquals(3, pubCount); + Assert.assertEquals(1, disconnectCount); + } + + private Future publishMsg() { + ByteBuf byteBuf = ALLOCATOR.buffer(); + byteBuf.writeBytes("payload".getBytes(StandardCharsets.UTF_8)); + return this.mqttClient.publish( + "test/topic", + byteBuf, + MqttQoS.AT_LEAST_ONCE); + } + + private MqttClient initClient() throws Exception { + MqttClientConfig config = new MqttClientConfig(); + config.setTimeoutSeconds(KEEPALIVE_TIMEOUT_SECONDS); + MqttClient client = MqttClient.create(config, null); + client.setEventLoop(this.eventLoopGroup); + Future 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 new file mode 100644 index 0000000000..602ca16911 --- /dev/null +++ b/netty-mqtt/src/test/java/org/thingsboard/mqtt/integration/server/MqttServer.java @@ -0,0 +1,84 @@ +/** + * Copyright © 2016-2022 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..."); + 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 new file mode 100644 index 0000000000..5c3db8da6a --- /dev/null +++ b/netty-mqtt/src/test/java/org/thingsboard/mqtt/integration/server/MqttTransportHandler.java @@ -0,0 +1,141 @@ +/** + * Copyright © 2016-2022 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); + } +}