Added test client to check and fix issue #21

2014-01-11 13:19:12 +01:00
parent a7619e889b
commit 1d9e2ccad2
5 changed files with 188 additions and 10 deletions
@@ -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());
+    }
+}
@@ -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<SocketChannel>() {
+                @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();
+    }
+}
@@ -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;
+    }
+    
+}