tsai
/
thingsboard
mirror of https://github.com/thingsboard/thingsboard.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

netty tests - added custom mqtt server and test with keepalive logic

pull/6967/head
Dima Landiak 4 years ago
parent
ac012d7c9e
commit
60ee2bb6ea
5 changed files with 396 additions and 0 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 5
      netty-mqtt/pom.xml
  2. 27
      netty-mqtt/src/test/java/org/thingsboard/mqtt/integration/IntegrationTestSuite.java
  3. 139
      netty-mqtt/src/test/java/org/thingsboard/mqtt/integration/MqttIntegrationTest.java
  4. 84
      netty-mqtt/src/test/java/org/thingsboard/mqtt/integration/server/MqttServer.java
  5. 141
      netty-mqtt/src/test/java/org/thingsboard/mqtt/integration/server/MqttTransportHandler.java

5
netty-mqtt/pom.xml
View File

@ -84,6 +84,11 @@
<artifactId>awaitility</artifactId> <artifactId>awaitility</artifactId>
<scope>test</scope> <scope>test</scope>
</dependency> </dependency>
<dependency>
<groupId>io.takari.junit</groupId>
<artifactId>takari-cpsuite</artifactId>
<scope>test</scope>
</dependency>
</dependencies> </dependencies>
<build> <build>

27
netty-mqtt/src/test/java/org/thingsboard/mqtt/integration/IntegrationTestSuite.java
View File

@ -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 {
}

139
netty-mqtt/src/test/java/org/thingsboard/mqtt/integration/MqttIntegrationTest.java
View File

@ -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<Void> 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<MqttMessageType> 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<Void> 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<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;
}
}

84
netty-mqtt/src/test/java/org/thingsboard/mqtt/integration/server/MqttServer.java
View File

@ -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<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...");
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!");
}
}

141
netty-mqtt/src/test/java/org/thingsboard/mqtt/integration/server/MqttTransportHandler.java
View File

@ -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<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);
}
}
Write Preview
Loading…
Cancel
Save