diff --git a/broker/src/test/java/org/eclipse/moquette/spec/v3_1_1/connection/ConnectionIT.java b/broker/src/test/java/org/eclipse/moquette/spec/v3_1_1/connection/ConnectionIT.java index 3a20ed64..bd68dd95 100644 --- a/broker/src/test/java/org/eclipse/moquette/spec/v3_1_1/connection/ConnectionIT.java +++ b/broker/src/test/java/org/eclipse/moquette/spec/v3_1_1/connection/ConnectionIT.java @@ -18,13 +18,12 @@ package org.eclipse.moquette.spec.v3_1_1.connection; import java.io.IOException; import org.eclipse.moquette.server.Server; +import org.eclipse.moquette.testclient.RawClient; import org.junit.After; import org.junit.Before; import org.junit.Rule; import org.junit.Ignore; import org.junit.Test; -import org.kaazing.robot.junit.annotation.Robotic; -import org.kaazing.robot.junit.rules.RobotRule; /** * @@ -33,9 +32,6 @@ import org.kaazing.robot.junit.rules.RobotRule; */ public class ConnectionIT { - @Rule - public RobotRule robot = new RobotRule(); - Server m_server; protected void startServer() throws IOException { @@ -52,30 +48,104 @@ public class ConnectionIT { public void tearDown() throws Exception { m_server.stopServer(); } - - // The test starts the server during setUp - // The network communication is driven by the script. - @Robotic(script = "connect.then.close") - @Test(timeout = 2000) - public void shouldConnectThenClose() throws Exception { - robot.join(); + + @Test(timeout = 3000) + public void testConnectThenClose() throws Exception { + RawClient.connect("127.0.0.1", 1883).isConnected() + //CONNECT + .write(0x10) //MQTT Control Packet type(1) + .write(0x13) // Remaining Length + .write(0x00, 0x04) // Protocol Name Length + .write("MQTT") // Protocol Name + .write(0x04) // The value of the Protocol Level field for the version 3.1.1 of the protocol is 4 (0x04) + + //Connect Flags + //User Name Flag(0) + //Password Flag(0) + //Will Retain(0) + //Will QoS(00) + //Will Flag(0) + //Clean Session(1) + .write(0x02) // Reserved(0) + .write(0x00, 0x00) // Keep Alive + + // Payload + .write(0x00, 0x07) // Client Identifier Length + .write("client1") // Client Identifier + .flush() + + //CONNACK + .read(0x20) // MQTT Control Packet type(2) + .read(0x02) // Remaining Length + + //Connect Acknowledge Flags + .read(0x00) // Session Present Flag(0) + + //Connect Return code + .read(0x00) // Connection Accepted + + //DISCONNECT + .write(0xE0) // MQTT Control Packet type(14) + .write(0x00) // Remaining Length + .closed(1000); } - - // TODO: server is not closing the connection immediately. - // The connection is closed 10 seconds after the CorruptedFrameException is - // thrown from ConnectDecoder - @Robotic(script = "connect.with.invalid.WillQoS") - @Test(timeout = 15000) - public void connectWithInvalidWillQoS() throws Exception { - robot.join(); + + @Test(timeout = 3000) + public void testConnectWithInvalidWillQoS() throws Exception { + RawClient.connect("127.0.0.1", 1883).isConnected() + //CONNECT + .write(0x10) //MQTT Control Packet type(1) + .write(0x13) // Remaining Length + .write(0x00, 0x04) // Protocol Name Length + .write("MQTT") // Protocol Name + .write(0x04) // The value of the Protocol Level field for the version 3.1.1 of the protocol is 4 (0x04) + + //Connect Flags + //User Name Flag(0) + //Password Flag(0) + //Will Retain(0) + //Will QoS(11) - It MUST NOT be 3 (0x03). Server should close the connection. + //Will Flag(1) + //Clean Session(1) + .write(0x1E) // Reserved(0) + .write(0x00, 0x00) // Keep Alive + + // Payload + .write(0x00, 0x07) // Client Identifier Length + .write("client1") // Client Identifier + .flush() + + .closed(1000); } - - + @Ignore("Need to validate the test case.") - @Robotic(script = "connect.with.WillFlag.set.to.zero.but.WillQoS.set.to.nonzero") @Test(timeout = 15000) - public void connectWithWillFlagSetToZeroButWillQoSFlagSetToNonZero() throws Exception { - robot.join(); + public void testConnectWithWillFlagSetToZeroButWillQoSFlagSetToNonZero() throws Exception { + RawClient.connect("127.0.0.1", 1883).isConnected() + // CONNECT + .write(0x10) // MQTT Control Packet type(1) + .write(0x12) // Remaining Length + .write(0x00, 0x04) // Protocol Name Length + .write("MQTT") // Protocol Name + .write(0x04) // The value of the Protocol Level field for the version 3.1.1 of the protocol is 4 (0x04) + + // Connect Flags + // User Name Flag(0) + // Password Flag(0) + // Will Retain(0) + // Will QoS(01) - If the Will Flag is set to 0, then the Will QoS MUST be set to 0 + // Will Flag(0) + //Clean Session(1) + .write(0x0A) // Reserved(0) + + .write(0x00, 0x00) // Keep Alive + + // Payload + .write(0x00, 0x07) // Client Identifier Length + .write("client1") //Client Identifier + + //Server MUST close the Network Connection due to invalid Will QoS flag + .closed(); } } diff --git a/broker/src/test/java/org/eclipse/moquette/testclient/RawClient.java b/broker/src/test/java/org/eclipse/moquette/testclient/RawClient.java new file mode 100644 index 00000000..af309b7d --- /dev/null +++ b/broker/src/test/java/org/eclipse/moquette/testclient/RawClient.java @@ -0,0 +1,167 @@ +package org.eclipse.moquette.testclient; + +import io.netty.bootstrap.Bootstrap; +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import io.netty.channel.*; +import io.netty.channel.nio.NioEventLoopGroup; +import io.netty.channel.socket.SocketChannel; +import io.netty.channel.socket.nio.NioSocketChannel; +import io.netty.util.ReferenceCountUtil; +import org.eclipse.moquette.parser.netty.Utils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.UnsupportedEncodingException; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; + +/** + * This class is used to have a fluent interface to interact with a server. + * Inspired by org.kaazing.robot + */ +public class RawClient { + + @ChannelHandler.Sharable + class RawMessageHandler extends ChannelInboundHandlerAdapter { + + @Override + public void channelRead(ChannelHandlerContext ctx, Object message) { + LOG.info("Received a message {}", message); + try { + ByteBuf in = (ByteBuf) message; + int readBytes = in.readableBytes(); + heapBuffer.writeBytes(in); + readableBytesSem.release(readBytes); + } finally { + ReferenceCountUtil.release(message); + } + } + + @Override + public void channelInactive(ChannelHandlerContext ctx) throws Exception { + //m_client.setConnectionLost(true); + disconnectLatch.countDown(); + ctx.close(/*false*/); + } + } + + private static final Logger LOG = LoggerFactory.getLogger(RawClient.class); + final RawMessageHandler handler; + EventLoopGroup workerGroup; + Channel m_channel; + + private boolean connected = false; + private ByteBuf heapBuffer; + private CountDownLatch disconnectLatch; + private final Semaphore readableBytesSem; + + private RawClient(String host, int port) { + handler = new RawMessageHandler(); + heapBuffer = Unpooled.buffer(128); + disconnectLatch = new CountDownLatch(1); + readableBytesSem = new Semaphore(0, true); + + 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("handler", handler); + } + }); + + // Start the client. + m_channel = b.connect(host, port).sync().channel(); + this.connected = true; + } catch (Exception ex) { + LOG.error("Error received in client setup", ex); + workerGroup.shutdownGracefully(); + } + } + + public static RawClient connect(String host, int port) { + return new RawClient(host, port); + } + + public RawClient isConnected() { + if (!this.connected) { + throw new IllegalStateException("Can't connect the client"); + } + return this; + } + + public RawClient write(int... bytes) { + ByteBuf buff = Unpooled.buffer(bytes.length); + buff.clear(); + for (int b : bytes) { + buff.writeByte((byte) b); + } + m_channel.write(buff); + return this; + } + + public RawClient writeWithSize(String str) { + ByteBuf buff = Unpooled.buffer(str.length() + 2); + buff.writeBytes(Utils.encodeString(str)); + m_channel.write(buff); + return this; + } + + /** + * Write just String bytes not length + * */ + public RawClient write(String str) { + ByteBuf out = Unpooled.buffer(str.length()); + byte[] raw; + try { + raw = str.getBytes("UTF-8"); + //NB every Java platform has got UTF-8 encoding by default, so this + //exception are never raised. + } catch (UnsupportedEncodingException ex) { + throw new IllegalStateException(ex); + } + out.writeBytes(raw); + m_channel.write(out); + return this; + } + + public RawClient flush() { + m_channel.flush(); + return this; + } + + public RawClient read(int expectedByte) { + try { + readableBytesSem.acquire(1); + } catch (InterruptedException e) { + throw new IllegalStateException("Interrupted while waiting data"); + } + byte b = heapBuffer.readByte(); + if (b != expectedByte) { + throw new IllegalStateException(String.format("Expected byte 0x%02X but found 0x%02X", b, expectedByte)); + } + return this; + } + + /** + * Expect the closing of the underling channel, with timeout + * */ + public void closed(long timeout) throws InterruptedException { + disconnectLatch.await(timeout, TimeUnit.MILLISECONDS); + } + + /** + * Expect closing the connect with unbound time + * */ + public void closed() throws InterruptedException { + disconnectLatch.await(); + } +} diff --git a/netty_parser/src/main/java/org/eclipse/moquette/parser/netty/Utils.java b/netty_parser/src/main/java/org/eclipse/moquette/parser/netty/Utils.java index 2b2f9123..66828e1c 100644 --- a/netty_parser/src/main/java/org/eclipse/moquette/parser/netty/Utils.java +++ b/netty_parser/src/main/java/org/eclipse/moquette/parser/netty/Utils.java @@ -136,7 +136,7 @@ public class Utils { * Return the IoBuffer with string encoded as MSB, LSB and UTF-8 encoded * string content. */ - static ByteBuf encodeString(String str) { + public static ByteBuf encodeString(String str) { ByteBuf out = Unpooled.buffer(2); byte[] raw; try {