anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Moved integration raw test from Kaazing robot to an internal Java DSL

Browse Source
This commit is contained in:
andrea
2015-03-04 00:26:09 +01:00
parent 50e1005b29
commit 528c263930
3 changed files with 263 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/test/java/org/eclipse/moquette/spec/v3_1_1/connection/ConnectionIT.java
+95 -25
View File
@@ -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();
}
}
broker/src/test/java/org/eclipse/moquette/testclient/RawClient.java
+167
View File
@@ -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<SocketChannel>() {
@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();
}
}