From 5f003ccb050faa1f5c0d5897c8705cde66cf9599 Mon Sep 17 00:00:00 2001 From: andsel Date: Mon, 2 May 2016 22:14:45 +0200 Subject: [PATCH] MOved conenct method into the test support class Client --- .../server/config/ConfigurationParser.java | 3 - ...erverLowlevelMessagesIntegrationTests.java | 65 +++------------- .../java/io/moquette/testclient/Client.java | 77 +++++++++++++++++-- .../proto/messages/AbstractMessage.java | 2 +- 4 files changed, 79 insertions(+), 68 deletions(-) diff --git a/broker/src/main/java/io/moquette/server/config/ConfigurationParser.java b/broker/src/main/java/io/moquette/server/config/ConfigurationParser.java index cc15e658..f44ea25b 100644 --- a/broker/src/main/java/io/moquette/server/config/ConfigurationParser.java +++ b/broker/src/main/java/io/moquette/server/config/ConfigurationParser.java @@ -37,9 +37,6 @@ class ConfigurationParser { private Properties m_properties = new Properties(); -// ConfigurationParser() { -// } - /** * Parse the configuration from file. */ diff --git a/broker/src/test/java/io/moquette/server/ServerLowlevelMessagesIntegrationTests.java b/broker/src/test/java/io/moquette/server/ServerLowlevelMessagesIntegrationTests.java index d0b161f2..2cbc05a7 100644 --- a/broker/src/test/java/io/moquette/server/ServerLowlevelMessagesIntegrationTests.java +++ b/broker/src/test/java/io/moquette/server/ServerLowlevelMessagesIntegrationTests.java @@ -31,8 +31,6 @@ import org.slf4j.LoggerFactory; import java.io.IOException; import java.util.Properties; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeUnit; import static org.junit.Assert.*; @@ -92,15 +90,15 @@ public class ServerLowlevelMessagesIntegrationTests { } @Test - public void checkWillMessageIsWiredOnClientKeepAliveExpiry() throws Exception { - LOG.info("*** checkWillMessageIsWiredOnClientKeepAliveExpiry ***"); + public void testWillMessageIsWiredOnClientKeepAliveExpiry() throws Exception { + LOG.info("*** testWillMessageIsWiredOnClientKeepAliveExpiry ***"); String willTestamentTopic = "/will/test"; String willTestamentMsg = "Bye bye"; m_willSubscriber.connect(); m_willSubscriber.subscribe(willTestamentTopic, 0); - connect(willTestamentTopic, willTestamentMsg); + m_client.clientId("FAKECLNT").connect(willTestamentTopic, willTestamentMsg); long connectTime = System.currentTimeMillis(); //but after the 2 KEEP ALIVE timeout expires it gets fired, @@ -116,30 +114,12 @@ public class ServerLowlevelMessagesIntegrationTests { } @Test - public void checkRejectConnectWithEmptyClientID() throws InterruptedException { - LOG.info("*** checkRejectConnectWithEmptyClientID ***"); - final CountDownLatch latch = new CountDownLatch(1); - m_client.setCallback(new Client.ICallback() { + public void testRejectConnectWithEmptyClientID() throws InterruptedException { + LOG.info("*** testRejectConnectWithEmptyClientID ***"); + m_client.clientId("").connect(); + + this.receivedMsg = this.m_client.lastReceivedMessage(); - public void call(AbstractMessage msg) { - receivedMsg = msg; - latch.countDown(); - } - }); - - int keepAlive = 2; //secs - ConnectMessage connectMessage = new ConnectMessage(); - connectMessage.setProtocolVersion((byte) 4); - connectMessage.setClientID(""); - connectMessage.setKeepAlive(keepAlive); - connectMessage.setWillFlag(false); - connectMessage.setWillQos(QOSType.MOST_ONE.byteValue()); - - //Execute - m_client.sendMessage(connectMessage); - - latch.await(200, TimeUnit.MILLISECONDS); - assertTrue(receivedMsg instanceof ConnAckMessage); ConnAckMessage connAck = (ConnAckMessage) receivedMsg; assertEquals(ConnAckMessage.IDENTIFIER_REJECTED, connAck.getReturnCode()); @@ -152,7 +132,7 @@ public class ServerLowlevelMessagesIntegrationTests { String willTestamentMsg = "Bye bye"; m_willSubscriber.connect(); m_willSubscriber.subscribe(willTestamentTopic, 0); - connect(willTestamentTopic, willTestamentMsg); + m_client.clientId("FAKECLNT").connect(willTestamentTopic, willTestamentMsg); //kill will publisher m_client.close(); @@ -163,31 +143,4 @@ public class ServerLowlevelMessagesIntegrationTests { m_willSubscriber.disconnect(); } - private void connect(String willTestamentTopic, String willTestamentMsg) throws InterruptedException { - final CountDownLatch latch = new CountDownLatch(1); - m_client.setCallback(new Client.ICallback() { - - public void call(AbstractMessage msg) { - receivedMsg = msg; - latch.countDown(); - } - }); - - int keepAlive = 2; //secs - ConnectMessage connectMessage = new ConnectMessage(); - connectMessage.setProtocolVersion((byte) 3); - connectMessage.setClientID("FAKECLNT"); - connectMessage.setKeepAlive(keepAlive); - connectMessage.setWillFlag(true); - connectMessage.setWillMessage(willTestamentMsg.getBytes()); - connectMessage.setWillTopic(willTestamentTopic); - connectMessage.setWillQos(QOSType.MOST_ONE.byteValue()); - - m_client.sendMessage(connectMessage); - - latch.await(200, TimeUnit.MILLISECONDS); - - assertTrue("conn-ack message is received", this.receivedMsg instanceof ConnAckMessage); - } - } diff --git a/broker/src/test/java/io/moquette/testclient/Client.java b/broker/src/test/java/io/moquette/testclient/Client.java index 5635b001..fa721f68 100644 --- a/broker/src/test/java/io/moquette/testclient/Client.java +++ b/broker/src/test/java/io/moquette/testclient/Client.java @@ -18,6 +18,8 @@ package io.moquette.testclient; import io.moquette.BrokerConstants; import io.moquette.parser.netty.MQTTDecoder; import io.moquette.parser.netty.MQTTEncoder; +import io.moquette.parser.proto.messages.ConnAckMessage; +import io.moquette.parser.proto.messages.ConnectMessage; import io.netty.bootstrap.Bootstrap; import io.netty.channel.Channel; import io.netty.channel.ChannelInitializer; @@ -31,6 +33,11 @@ import io.moquette.parser.proto.messages.AbstractMessage; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import static org.junit.Assert.assertTrue; + /** * Class used just to send and receive MQTT messages without any protocol login * in action, just use the encoder/decoder part. @@ -39,7 +46,7 @@ import org.slf4j.LoggerFactory; */ public class Client { - public static interface ICallback { + public interface ICallback { void call(AbstractMessage msg); } @@ -48,17 +55,17 @@ public class Client { EventLoopGroup workerGroup; Channel m_channel; private boolean m_connectionLost = false; + private ICallback callback; + private String clientId; + private AbstractMessage receivedMsg; public Client(String host) { this(host, BrokerConstants.PORT); } public Client(String host, int port) { - handler.setClient(this); - workerGroup = new NioEventLoopGroup(); - try { Bootstrap b = new Bootstrap(); b.group(workerGroup); @@ -81,8 +88,58 @@ public class Client { workerGroup.shutdownGracefully(); } } - - ICallback callback; + + public Client clientId(String clientId) { + this.clientId = clientId; + return this; + } + + public void connect(String willTestamentTopic, String willTestamentMsg) { + ConnectMessage connectMessage = new ConnectMessage(); + connectMessage.setProtocolVersion((byte) 3); + connectMessage.setClientID(this.clientId); + connectMessage.setKeepAlive(2); //secs + connectMessage.setWillFlag(true); + connectMessage.setWillMessage(willTestamentMsg.getBytes()); + connectMessage.setWillTopic(willTestamentTopic); + connectMessage.setWillQos(AbstractMessage.QOSType.MOST_ONE.byteValue()); + + doConnect(connectMessage); + } + + public void connect() { + ConnectMessage connectMessage = new ConnectMessage(); + connectMessage.setProtocolVersion((byte) 4); + connectMessage.setClientID(""); + connectMessage.setKeepAlive(2); //secs + connectMessage.setWillFlag(false); + connectMessage.setWillQos(AbstractMessage.QOSType.MOST_ONE.byteValue()); + + doConnect(connectMessage); + } + + private void doConnect(ConnectMessage connectMessage) { + final CountDownLatch latch = new CountDownLatch(1); + this.setCallback(new Client.ICallback() { + + public void call(AbstractMessage msg) { + receivedMsg = msg; + latch.countDown(); + } + }); + + this.sendMessage(connectMessage); + + try { + latch.await(200, TimeUnit.MILLISECONDS); + } catch (InterruptedException e) { + throw new RuntimeException("Cannot receive message in 200 ms", e); + } + if (!(this.receivedMsg instanceof ConnAckMessage)) { + throw new RuntimeException("Expected a CONN_ACK message but received " + this.receivedMsg.getMessageType()); + } + } + public void setCallback(ICallback callback) { this.callback = callback; } @@ -91,8 +148,12 @@ public class Client { m_channel.writeAndFlush(msg); } + public AbstractMessage lastReceivedMessage() { + return this.receivedMsg; + } + void messageReceived(AbstractMessage msg) { - LOG.info("Received message " + msg); + LOG.info("Received message {}", msg); if (this.callback != null) { this.callback.call(msg); } @@ -105,7 +166,7 @@ public class Client { public boolean isConnectionLost() { return m_connectionLost; } - + public void close() throws InterruptedException { // Wait until the connection is closed. m_channel.closeFuture().sync(); diff --git a/netty_parser/src/main/java/io/moquette/parser/proto/messages/AbstractMessage.java b/netty_parser/src/main/java/io/moquette/parser/proto/messages/AbstractMessage.java index dab95ee7..34cad5d1 100644 --- a/netty_parser/src/main/java/io/moquette/parser/proto/messages/AbstractMessage.java +++ b/netty_parser/src/main/java/io/moquette/parser/proto/messages/AbstractMessage.java @@ -37,7 +37,7 @@ public abstract class AbstractMessage { public static final byte PINGRESP = 13; //PING Response public static final byte DISCONNECT = 14; //Client is Disconnecting - public static enum QOSType { + public enum QOSType { MOST_ONE, LEAST_ONE, EXACTLY_ONCE, RESERVED, FAILURE; public static QOSType valueOf(byte qos){