diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/Server.java b/broker/src/main/java/org/dna/mqtt/moquette/server/Server.java index 98830f8d..3a520e43 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/server/Server.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/server/Server.java @@ -2,9 +2,7 @@ package org.dna.mqtt.moquette.server; import java.io.File; import java.io.IOException; -import java.util.logging.Level; import org.dna.mqtt.moquette.messaging.spi.impl.SimpleMessaging; -//import org.dna.mqtt.moquette.server.mina.MinaAcceptor; import org.dna.mqtt.moquette.server.netty.NettyAcceptor; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -50,11 +48,5 @@ public class Server { messaging.stop(); m_acceptor.close(); System.out.println("Server stopped"); -// try { -// //sleep one second to give the disruptor -// Thread.sleep(100); -// } catch (InterruptedException ex) { -// LOG.error(null, ex); -// } } } diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/netty/MoquetteIdleTimoutHandler.java b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/MoquetteIdleTimoutHandler.java index a02de9d4..6e7de621 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/server/netty/MoquetteIdleTimoutHandler.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/MoquetteIdleTimoutHandler.java @@ -3,12 +3,13 @@ package org.dna.mqtt.moquette.server.netty; import io.netty.channel.ChannelDuplexHandler; import io.netty.channel.ChannelHandlerContext; import io.netty.handler.timeout.IdleState; +import io.netty.handler.timeout.IdleStateEvent; class MoquetteIdleTimoutHandler extends ChannelDuplexHandler { @Override public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { - if (evt instanceof IdleState) { - IdleState e = (IdleState) evt; + if (evt instanceof IdleStateEvent) { + IdleState e = ((IdleStateEvent) evt).state(); if (e == IdleState.ALL_IDLE) { ctx.close(); } /*else if (e.getState() == IdleState.WRITER_IDLE) { diff --git a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerLowlevelMessagesIntegrationTests.java b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerLowlevelMessagesIntegrationTests.java new file mode 100644 index 00000000..1c5ee7c3 --- /dev/null +++ b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerLowlevelMessagesIntegrationTests.java @@ -0,0 +1,56 @@ +package org.dna.mqtt.moquette.server; + +import java.io.File; +import java.io.IOException; +import org.dna.mqtt.moquette.proto.messages.ConnectMessage; +import org.dna.mqtt.moquette.testclient.Client; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import static org.junit.Assert.*; + +/** + * + * @author andrea + */ +public class ServerLowlevelMessagesIntegrationTests { + Server m_server; + Client m_client; + + protected void startServer() throws IOException { + m_server = new Server(); + m_server.startServer(); + } + + @Before + public void setUp() throws Exception { + startServer(); + m_client = new Client("localhost"); + } + + @After + public void tearDown() throws Exception { + m_client.close(); + + m_server.stopServer(); + File dbFile = new File(Server.STORAGE_FILE_PATH); + if (dbFile.exists()) { + dbFile.delete(); + } + } + + @Test + public void elapseKeepAliveTime() throws InterruptedException { + int keepAlive = 2; //secs + ConnectMessage connectMessage = new ConnectMessage(); + connectMessage.setProcotolVersion((byte)3); + connectMessage.setClientID("FAKECLNT"); + connectMessage.setKeepAlive(keepAlive); + m_client.sendMessage(connectMessage); + + //wait 2 times the keepAlive + Thread.sleep(keepAlive * 2 * 1000); + + assertTrue(m_client.isConnectionLost()); + } +} diff --git a/broker/src/test/java/org/dna/mqtt/moquette/testclient/Client.java b/broker/src/test/java/org/dna/mqtt/moquette/testclient/Client.java new file mode 100644 index 00000000..4d33b38e --- /dev/null +++ b/broker/src/test/java/org/dna/mqtt/moquette/testclient/Client.java @@ -0,0 +1,90 @@ +package org.dna.mqtt.moquette.testclient; + +import io.netty.bootstrap.Bootstrap; +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.NioSocketChannel; +import org.dna.mqtt.commons.Constants; +import org.dna.mqtt.moquette.parser.netty.MQTTDecoder; +import org.dna.mqtt.moquette.parser.netty.MQTTEncoder; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Class used just to send and receive MQTT messages without any protocol login + * in action, just use the encoder/decoder part. + * + * @author andrea + */ +public class Client { + + private static final Logger LOG = LoggerFactory.getLogger(Client.class); + final ClientNettyMQTTHandler handler = new ClientNettyMQTTHandler(); + EventLoopGroup workerGroup; + Channel m_channel; + private boolean m_connectionLost = false; + + public Client(String host) { + this(host, Constants.PORT); + } + + public Client(String host, int port) { + + handler.setClient(this); + + workerGroup = new NioEventLoopGroup(); + + try { + Bootstrap b = new Bootstrap(); + b.group(workerGroup); + b.channel(NioSocketChannel.class); + b.option(ChannelOption.SO_KEEPALIVE, true); + b.handler(new ChannelInitializer() { + @Override + public void initChannel(SocketChannel ch) throws Exception { + ChannelPipeline pipeline = ch.pipeline(); + pipeline.addLast("decoder", new MQTTDecoder()); + pipeline.addLast("encoder", new MQTTEncoder()); + pipeline.addLast("handler", handler); + } + }); + + // Start the client. + m_channel = b.connect(host, port).sync().channel(); + } catch (Exception ex) { + LOG.error("Error received in client setup", ex); + workerGroup.shutdownGracefully(); + } + } + + public void sendMessage(AbstractMessage msg) { + m_channel.writeAndFlush(msg); + } + + void messageReceived(AbstractMessage msg) { + LOG.info("Received message " + msg); + } + + void setConnectionLost(boolean status) { + m_connectionLost = status; + } + + public boolean isConnectionLost() { + return m_connectionLost; + } + + public void close() throws InterruptedException { + // Wait until the connection is closed. + m_channel.closeFuture().sync(); + if (workerGroup == null) { + throw new IllegalStateException("Invoked close on an Acceptor that wasn't initialized"); + } + workerGroup.shutdownGracefully(); + } +} diff --git a/broker/src/test/java/org/dna/mqtt/moquette/testclient/ClientNettyMQTTHandler.java b/broker/src/test/java/org/dna/mqtt/moquette/testclient/ClientNettyMQTTHandler.java new file mode 100644 index 00000000..9a802907 --- /dev/null +++ b/broker/src/test/java/org/dna/mqtt/moquette/testclient/ClientNettyMQTTHandler.java @@ -0,0 +1,39 @@ +package org.dna.mqtt.moquette.testclient; + +import io.netty.channel.ChannelHandler; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelInboundHandlerAdapter; +import org.dna.mqtt.moquette.proto.Utils; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.server.netty.NettyMQTTHandler; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * + * @author andrea + */ +@ChannelHandler.Sharable +class ClientNettyMQTTHandler extends ChannelInboundHandlerAdapter { + + private static final Logger LOG = LoggerFactory.getLogger(NettyMQTTHandler.class); + private Client m_client; + + @Override + public void channelRead(ChannelHandlerContext ctx, Object message) { + AbstractMessage msg = (AbstractMessage) message; + LOG.info("Received a message of type {}", Utils.msgType2String(msg.getMessageType())); + m_client.messageReceived(msg); + } + + @Override + public void channelInactive(ChannelHandlerContext ctx)throws Exception { + m_client.setConnectionLost(true); + ctx.close(/*false*/); + } + + void setClient(Client client) { + m_client = client; + } + +}