From ea62a684d0a43f44f4aef3da870c0558bce6d800 Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Wed, 12 Sep 2012 00:13:24 +0200 Subject: [PATCH] Completed the exporting into broker module --- .../mqtt/bechnmark/mina/MQTTBulkClient.java | 4 +- broker/pom.xml | 147 ++++++++++++++++++ .../spi/IMatchingCondition.java | 0 .../{ => messaging}/spi/IMessaging.java | 0 .../{ => messaging}/spi/INotifier.java | 0 .../{ => messaging}/spi/IStorageService.java | 0 .../spi/impl/HawtDBStorageService.java | 1 + .../spi/impl/SimpleMessaging.java | 0 .../{ => messaging}/spi/impl/ValueEvent.java | 0 .../spi/impl/events/DisconnectEvent.java | 0 .../spi/impl/events/InitEvent.java | 0 .../spi/impl/events/MessagingEvent.java | 0 .../spi/impl/events/NotifyEvent.java | 0 .../spi/impl/events/ProtocolEvent.java | 0 .../spi/impl/events/PubAckEvent.java | 0 .../spi/impl/events/PublishEvent.java | 0 .../spi/impl/events/RepublishEvent.java | 0 .../spi/impl/events/StopEvent.java | 0 .../spi/impl/events/SubscribeEvent.java | 0 .../spi/impl/subscriptions/Subscription.java | 0 .../subscriptions/SubscriptionsStore.java | 0 .../spi/impl/subscriptions/Token.java | 0 .../spi/impl/subscriptions/TreeNode.java | 0 .../org/dna/mqtt/moquette/server/Server.java | 5 +- .../impl/DummyStorageService.java | 0 .../impl/SimpleMessagingTest.java | 0 .../subscriptions/SubscriptionsStoreTest.java | 0 .../server/ServerIntegrationFuseTest.java | 5 +- .../server/ServerIntegrationTest.java | 29 ++-- parser/pom.xml | 13 +- .../java/org/dna/mqtt/commons/Constants.java | 9 ++ .../org/dna/mqtt/moquette/client/Client.java | 4 +- pom.xml | 1 + 33 files changed, 189 insertions(+), 29 deletions(-) create mode 100644 broker/pom.xml rename broker/src/main/java/org/dna/mqtt/moquette/{ => messaging}/spi/IMatchingCondition.java (100%) rename broker/src/main/java/org/dna/mqtt/moquette/{ => messaging}/spi/IMessaging.java (100%) rename broker/src/main/java/org/dna/mqtt/moquette/{ => messaging}/spi/INotifier.java (100%) rename broker/src/main/java/org/dna/mqtt/moquette/{ => messaging}/spi/IStorageService.java (100%) rename broker/src/main/java/org/dna/mqtt/moquette/{ => messaging}/spi/impl/HawtDBStorageService.java (99%) rename broker/src/main/java/org/dna/mqtt/moquette/{ => messaging}/spi/impl/SimpleMessaging.java (100%) rename broker/src/main/java/org/dna/mqtt/moquette/{ => messaging}/spi/impl/ValueEvent.java (100%) rename broker/src/main/java/org/dna/mqtt/moquette/{ => messaging}/spi/impl/events/DisconnectEvent.java (100%) rename broker/src/main/java/org/dna/mqtt/moquette/{ => messaging}/spi/impl/events/InitEvent.java (100%) rename broker/src/main/java/org/dna/mqtt/moquette/{ => messaging}/spi/impl/events/MessagingEvent.java (100%) rename broker/src/main/java/org/dna/mqtt/moquette/{ => messaging}/spi/impl/events/NotifyEvent.java (100%) rename broker/src/main/java/org/dna/mqtt/moquette/{ => messaging}/spi/impl/events/ProtocolEvent.java (100%) rename broker/src/main/java/org/dna/mqtt/moquette/{ => messaging}/spi/impl/events/PubAckEvent.java (100%) rename broker/src/main/java/org/dna/mqtt/moquette/{ => messaging}/spi/impl/events/PublishEvent.java (100%) rename broker/src/main/java/org/dna/mqtt/moquette/{ => messaging}/spi/impl/events/RepublishEvent.java (100%) rename broker/src/main/java/org/dna/mqtt/moquette/{ => messaging}/spi/impl/events/StopEvent.java (100%) rename broker/src/main/java/org/dna/mqtt/moquette/{ => messaging}/spi/impl/events/SubscribeEvent.java (100%) rename broker/src/main/java/org/dna/mqtt/moquette/{ => messaging}/spi/impl/subscriptions/Subscription.java (100%) rename broker/src/main/java/org/dna/mqtt/moquette/{ => messaging}/spi/impl/subscriptions/SubscriptionsStore.java (100%) rename broker/src/main/java/org/dna/mqtt/moquette/{ => messaging}/spi/impl/subscriptions/Token.java (100%) rename broker/src/main/java/org/dna/mqtt/moquette/{ => messaging}/spi/impl/subscriptions/TreeNode.java (100%) rename broker/src/test/java/org/dna/mqtt/moquette/{ => messaging}/impl/DummyStorageService.java (100%) rename broker/src/test/java/org/dna/mqtt/moquette/{ => messaging}/impl/SimpleMessagingTest.java (100%) rename broker/src/test/java/org/dna/mqtt/moquette/{ => messaging}/impl/subscriptions/SubscriptionsStoreTest.java (100%) create mode 100644 parser/src/main/java/org/dna/mqtt/commons/Constants.java diff --git a/benchmarking/src/main/java/org/dna/mqtt/bechnmark/mina/MQTTBulkClient.java b/benchmarking/src/main/java/org/dna/mqtt/bechnmark/mina/MQTTBulkClient.java index bf2728de..4def7701 100644 --- a/benchmarking/src/main/java/org/dna/mqtt/bechnmark/mina/MQTTBulkClient.java +++ b/benchmarking/src/main/java/org/dna/mqtt/bechnmark/mina/MQTTBulkClient.java @@ -12,11 +12,11 @@ import org.apache.mina.filter.codec.ProtocolCodecFilter; import org.apache.mina.filter.codec.demux.DemuxingProtocolDecoder; import org.apache.mina.filter.codec.demux.DemuxingProtocolEncoder; import org.apache.mina.transport.socket.nio.NioSocketConnector; +import org.dna.mqtt.commons.Constants; import org.dna.mqtt.moquette.MQTTException; import org.dna.mqtt.moquette.PublishException; import org.dna.mqtt.moquette.proto.*; import org.dna.mqtt.moquette.proto.messages.*; -import org.dna.mqtt.moquette.server.Server; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -81,7 +81,7 @@ public class MQTTBulkClient { m_connector.setHandler(new DummyClientHandler()); m_connector.getSessionConfig().setReadBufferSize(2048); - m_connector.getSessionConfig().setIdleTime(IdleStatus.BOTH_IDLE, Server.DEFAULT_CONNECT_TIMEOUT); + m_connector.getSessionConfig().setIdleTime(IdleStatus.BOTH_IDLE, Constants.DEFAULT_CONNECT_TIMEOUT); } public void connect() throws MQTTException { diff --git a/broker/pom.xml b/broker/pom.xml new file mode 100644 index 00000000..449577b2 --- /dev/null +++ b/broker/pom.xml @@ -0,0 +1,147 @@ + + + 4.0.0 + + ../pom.xml + moquette-parent + org.dna.mqtt + 0.1 + + + moquette-broker + jar + 0.1 + Moquette - broker + + + UTF-8 + + + + + org.dna.mqtt + moquette-parser + 0.1 + + + + org.apache.mina + mina-core + 2.0.4 + + + + com.googlecode.disruptor + disruptor + 2.7.1 + + + + + + + org.slf4j + slf4j-log4j12 + 1.6.4 + + + + org.fusesource.hawtdb + hawtdb + 1.6 + + + + junit + junit + 4.8.2 + test + + + + + org.mockito + mockito-core + 1.9.0 + test + + + + org.fusesource.mqtt-client + mqtt-client + + 1.3 + test + + + + org.fusesource.hawtbuf + hawtbuf + 1.9 + + + + + + + org.apache.maven.plugins + maven-compiler-plugin + + 1.5 + 1.5 + + + + + org.apache.maven.plugins + maven-surefire-plugin + + + org/dna/mqtt/moquette/proto/TestUtils.java + org/dna/mqtt/moquette/proto/TestUtils$* + + + + + + + maven-assembly-plugin + + + + org.dna.mqtt.moquette.server.Server + + + + jar-with-dependencies + + + + + + + + + + + + + diff --git a/broker/src/main/java/org/dna/mqtt/moquette/spi/IMatchingCondition.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMatchingCondition.java similarity index 100% rename from broker/src/main/java/org/dna/mqtt/moquette/spi/IMatchingCondition.java rename to broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMatchingCondition.java diff --git a/broker/src/main/java/org/dna/mqtt/moquette/spi/IMessaging.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java similarity index 100% rename from broker/src/main/java/org/dna/mqtt/moquette/spi/IMessaging.java rename to broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java diff --git a/broker/src/main/java/org/dna/mqtt/moquette/spi/INotifier.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/INotifier.java similarity index 100% rename from broker/src/main/java/org/dna/mqtt/moquette/spi/INotifier.java rename to broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/INotifier.java diff --git a/broker/src/main/java/org/dna/mqtt/moquette/spi/IStorageService.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java similarity index 100% rename from broker/src/main/java/org/dna/mqtt/moquette/spi/IStorageService.java rename to broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java diff --git a/broker/src/main/java/org/dna/mqtt/moquette/spi/impl/HawtDBStorageService.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java similarity index 99% rename from broker/src/main/java/org/dna/mqtt/moquette/spi/impl/HawtDBStorageService.java rename to broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java index 09408816..88e1386e 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/spi/impl/HawtDBStorageService.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java @@ -15,6 +15,7 @@ import org.slf4j.LoggerFactory; import java.io.File; import java.io.IOException; import java.io.Serializable; +import java.util.*; /** diff --git a/broker/src/main/java/org/dna/mqtt/moquette/spi/impl/SimpleMessaging.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java similarity index 100% rename from broker/src/main/java/org/dna/mqtt/moquette/spi/impl/SimpleMessaging.java rename to broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java diff --git a/broker/src/main/java/org/dna/mqtt/moquette/spi/impl/ValueEvent.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ValueEvent.java similarity index 100% rename from broker/src/main/java/org/dna/mqtt/moquette/spi/impl/ValueEvent.java rename to broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ValueEvent.java diff --git a/broker/src/main/java/org/dna/mqtt/moquette/spi/impl/events/DisconnectEvent.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/DisconnectEvent.java similarity index 100% rename from broker/src/main/java/org/dna/mqtt/moquette/spi/impl/events/DisconnectEvent.java rename to broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/DisconnectEvent.java diff --git a/broker/src/main/java/org/dna/mqtt/moquette/spi/impl/events/InitEvent.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/InitEvent.java similarity index 100% rename from broker/src/main/java/org/dna/mqtt/moquette/spi/impl/events/InitEvent.java rename to broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/InitEvent.java diff --git a/broker/src/main/java/org/dna/mqtt/moquette/spi/impl/events/MessagingEvent.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/MessagingEvent.java similarity index 100% rename from broker/src/main/java/org/dna/mqtt/moquette/spi/impl/events/MessagingEvent.java rename to broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/MessagingEvent.java diff --git a/broker/src/main/java/org/dna/mqtt/moquette/spi/impl/events/NotifyEvent.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/NotifyEvent.java similarity index 100% rename from broker/src/main/java/org/dna/mqtt/moquette/spi/impl/events/NotifyEvent.java rename to broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/NotifyEvent.java diff --git a/broker/src/main/java/org/dna/mqtt/moquette/spi/impl/events/ProtocolEvent.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/ProtocolEvent.java similarity index 100% rename from broker/src/main/java/org/dna/mqtt/moquette/spi/impl/events/ProtocolEvent.java rename to broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/ProtocolEvent.java diff --git a/broker/src/main/java/org/dna/mqtt/moquette/spi/impl/events/PubAckEvent.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PubAckEvent.java similarity index 100% rename from broker/src/main/java/org/dna/mqtt/moquette/spi/impl/events/PubAckEvent.java rename to broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PubAckEvent.java diff --git a/broker/src/main/java/org/dna/mqtt/moquette/spi/impl/events/PublishEvent.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java similarity index 100% rename from broker/src/main/java/org/dna/mqtt/moquette/spi/impl/events/PublishEvent.java rename to broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java diff --git a/broker/src/main/java/org/dna/mqtt/moquette/spi/impl/events/RepublishEvent.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/RepublishEvent.java similarity index 100% rename from broker/src/main/java/org/dna/mqtt/moquette/spi/impl/events/RepublishEvent.java rename to broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/RepublishEvent.java diff --git a/broker/src/main/java/org/dna/mqtt/moquette/spi/impl/events/StopEvent.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/StopEvent.java similarity index 100% rename from broker/src/main/java/org/dna/mqtt/moquette/spi/impl/events/StopEvent.java rename to broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/StopEvent.java diff --git a/broker/src/main/java/org/dna/mqtt/moquette/spi/impl/events/SubscribeEvent.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/SubscribeEvent.java similarity index 100% rename from broker/src/main/java/org/dna/mqtt/moquette/spi/impl/events/SubscribeEvent.java rename to broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/SubscribeEvent.java diff --git a/broker/src/main/java/org/dna/mqtt/moquette/spi/impl/subscriptions/Subscription.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/Subscription.java similarity index 100% rename from broker/src/main/java/org/dna/mqtt/moquette/spi/impl/subscriptions/Subscription.java rename to broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/Subscription.java diff --git a/broker/src/main/java/org/dna/mqtt/moquette/spi/impl/subscriptions/SubscriptionsStore.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStore.java similarity index 100% rename from broker/src/main/java/org/dna/mqtt/moquette/spi/impl/subscriptions/SubscriptionsStore.java rename to broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStore.java diff --git a/broker/src/main/java/org/dna/mqtt/moquette/spi/impl/subscriptions/Token.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/Token.java similarity index 100% rename from broker/src/main/java/org/dna/mqtt/moquette/spi/impl/subscriptions/Token.java rename to broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/Token.java diff --git a/broker/src/main/java/org/dna/mqtt/moquette/spi/impl/subscriptions/TreeNode.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/TreeNode.java similarity index 100% rename from broker/src/main/java/org/dna/mqtt/moquette/spi/impl/subscriptions/TreeNode.java rename to broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/TreeNode.java 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 79d9280d..9f8624e6 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 @@ -15,6 +15,7 @@ import org.apache.mina.transport.socket.nio.NioSocketAcceptor; import org.dna.mqtt.moquette.messaging.spi.impl.SimpleMessaging; import org.dna.mqtt.moquette.proto.*; import org.dna.mqtt.moquette.proto.messages.*; +import org.dna.mqtt.commons.Constants; import org.slf4j.Logger; import org.slf4j.LoggerFactory; /** @@ -86,10 +87,10 @@ public class Server { ((NioSocketAcceptor)m_acceptor).setReuseAddress(true); ((NioSocketAcceptor)m_acceptor).getSessionConfig().setReuseAddress(true); m_acceptor.getSessionConfig().setReadBufferSize( 2048 ); - m_acceptor.getSessionConfig().setIdleTime( IdleStatus.BOTH_IDLE, DEFAULT_CONNECT_TIMEOUT ); + m_acceptor.getSessionConfig().setIdleTime( IdleStatus.BOTH_IDLE, Constants.DEFAULT_CONNECT_TIMEOUT ); m_acceptor.getStatistics().setThroughputCalculationInterval(10); m_acceptor.getStatistics().updateThroughput(System.currentTimeMillis()); - m_acceptor.bind( new InetSocketAddress(PORT) ); + m_acceptor.bind( new InetSocketAddress(Constants.PORT) ); LOG.info("Server binded"); //Bind a shutdown hook diff --git a/broker/src/test/java/org/dna/mqtt/moquette/impl/DummyStorageService.java b/broker/src/test/java/org/dna/mqtt/moquette/messaging/impl/DummyStorageService.java similarity index 100% rename from broker/src/test/java/org/dna/mqtt/moquette/impl/DummyStorageService.java rename to broker/src/test/java/org/dna/mqtt/moquette/messaging/impl/DummyStorageService.java diff --git a/broker/src/test/java/org/dna/mqtt/moquette/impl/SimpleMessagingTest.java b/broker/src/test/java/org/dna/mqtt/moquette/messaging/impl/SimpleMessagingTest.java similarity index 100% rename from broker/src/test/java/org/dna/mqtt/moquette/impl/SimpleMessagingTest.java rename to broker/src/test/java/org/dna/mqtt/moquette/messaging/impl/SimpleMessagingTest.java diff --git a/broker/src/test/java/org/dna/mqtt/moquette/impl/subscriptions/SubscriptionsStoreTest.java b/broker/src/test/java/org/dna/mqtt/moquette/messaging/impl/subscriptions/SubscriptionsStoreTest.java similarity index 100% rename from broker/src/test/java/org/dna/mqtt/moquette/impl/subscriptions/SubscriptionsStoreTest.java rename to broker/src/test/java/org/dna/mqtt/moquette/messaging/impl/subscriptions/SubscriptionsStoreTest.java diff --git a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationFuseTest.java b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationFuseTest.java index 71286d8f..4ffbedc5 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationFuseTest.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationFuseTest.java @@ -1,6 +1,7 @@ package org.dna.mqtt.moquette.server; import org.fusesource.mqtt.client.*; +import org.dna.mqtt.commons.Constants; import org.junit.After; import org.junit.Before; import org.junit.Test; @@ -30,7 +31,7 @@ public class ServerIntegrationFuseTest { startServer(); m_mqtt = new MQTT(); - m_mqtt.setHost("localhost", Server.PORT); + m_mqtt.setHost("localhost", Constants.PORT); m_mqtt.setClientId("TestClient"); } @@ -113,7 +114,7 @@ public class ServerIntegrationFuseTest { private void publishFromAnotherClient(String topic, byte[] payload, QoS qos) throws Exception { MQTT mqttPub = new MQTT(); - mqttPub.setHost("localhost", Server.PORT); + mqttPub.setHost("localhost", Constants.PORT); mqttPub.setClientId("TestClientPUB"); FutureConnection connectionPub = mqttPub.futureConnection(); connectionPub.connect().await(); diff --git a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java index f6fce7d0..3b73eb82 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java @@ -7,6 +7,7 @@ import java.util.concurrent.TimeUnit; import org.dna.mqtt.moquette.client.Client; import org.dna.mqtt.moquette.client.IPublishCallback; import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.commons.Constants; import org.junit.After; import static org.junit.Assert.*; import org.junit.Before; @@ -44,7 +45,7 @@ public class ServerIntegrationTest { @Test public void testSubscribe() throws IOException, InterruptedException { // startServer(); - Client client = new Client("localhost", Server.PORT); + Client client = new Client("localhost", Constants.PORT); // Client client = new Client("test.mosquitto.org", 1883); client.connect(); @@ -67,7 +68,7 @@ public class ServerIntegrationTest { @Test public void testCleanSession_maintainClientSubscriptions() { - Client client = new Client("localhost", Server.PORT/*, "CLID_123"*/); + Client client = new Client("localhost", Constants.PORT/*, "CLID_123"*/); client.connect(false); //without session cleanup client.subscribe("/topic", new IPublishCallback() { @@ -92,7 +93,7 @@ public class ServerIntegrationTest { @Test public void testCleanSession_maintainClientSubscriptions_againstClientDestruction() { - Client client = new Client("localhost", Server.PORT, "CLID_123"); + Client client = new Client("localhost", Constants.PORT, "CLID_123"); client.connect(false); //without session cleanup client.subscribe("/topic", new IPublishCallback() { @@ -105,7 +106,7 @@ public class ServerIntegrationTest { client.shutdown(); //Exercise that the client maintain the subscriptions - client = new Client("localhost", Server.PORT, "CLID_123"); + client = new Client("localhost", Constants.PORT, "CLID_123"); client.register("/topic", new IPublishCallback() { public void published(String topic, byte[] message) { @@ -131,7 +132,7 @@ public class ServerIntegrationTest { */ @Test public void testCleanSession_correctlyClientSubscriptions() { - Client client = new Client("localhost", Server.PORT, "CLID_123"); + Client client = new Client("localhost", Constants.PORT, "CLID_123"); client.connect(false); //without session cleanup client.subscribe("/topic", new IPublishCallback() { @@ -144,7 +145,7 @@ public class ServerIntegrationTest { client.shutdown(); //Exercise that the client maintain the subscriptions - client = new Client("localhost", Server.PORT, "CLID_123"); + client = new Client("localhost", Constants.PORT, "CLID_123"); client.register("/topic", new IPublishCallback() { public void published(String topic, byte[] message) { @@ -166,7 +167,7 @@ public class ServerIntegrationTest { @Test public void testCleanSession_maintainClientSubscriptions_withServerRestart() throws IOException, InterruptedException { final CountDownLatch barrier = new CountDownLatch(1); - Client client = new Client("localhost", Server.PORT, "CLID_123"); + Client client = new Client("localhost", Constants.PORT, "CLID_123"); client.connect(false); //without session cleanup client.subscribe("/topic", new IPublishCallback() { @@ -188,7 +189,7 @@ public class ServerIntegrationTest { final CountDownLatch barrier2 = new CountDownLatch(1); //Exercise that the client maintain the subscriptions - client = new Client("localhost", Server.PORT, "CLID_123"); + client = new Client("localhost", Constants.PORT, "CLID_123"); client.register("/topic", new IPublishCallback() { public void published(String topic, byte[] message) { @@ -212,7 +213,7 @@ public class ServerIntegrationTest { @Test public void testRetain_maintainMessage_againstClientDestruction() throws InterruptedException { final CountDownLatch barrier = new CountDownLatch(1); - Client client = new Client("localhost", Server.PORT, "CLID_123"); + Client client = new Client("localhost", Constants.PORT, "CLID_123"); client.connect(); client.publish("/topic", "Test my payload".getBytes(), true); @@ -220,7 +221,7 @@ public class ServerIntegrationTest { client.shutdown(); //Exercise that the client maintain the subscriptions - client = new Client("localhost", Server.PORT, "CLID_123"); + client = new Client("localhost", Constants.PORT, "CLID_123"); client.connect(); client.subscribe("/topic", new IPublishCallback() { @@ -242,7 +243,7 @@ public class ServerIntegrationTest { @Ignore public void testUnsubscribe_do_not_notify_anymore_same_session() throws InterruptedException { - Client client = new Client("localhost", Server.PORT, "CLID_123"); + Client client = new Client("localhost", Constants.PORT, "CLID_123"); // Client client = new Client("test.mosquitto.org", 1883); client.connect(); @@ -276,7 +277,7 @@ public class ServerIntegrationTest { @Test public void testUnsubscribe_do_not_notify_anymore_new_session() throws InterruptedException { - Client client = new Client("localhost", Server.PORT, "CLID_123"); + Client client = new Client("localhost", Constants.PORT, "CLID_123"); // Client client = new Client("test.mosquitto.org", 1883); client.connect(); @@ -304,7 +305,7 @@ public class ServerIntegrationTest { client.shutdown(); //Exercise that the client maintain the subscriptions - client = new Client("localhost", Server.PORT, "CLID_123"); + client = new Client("localhost", Constants.PORT, "CLID_123"); client.connect(); //reinit the flag @@ -324,7 +325,7 @@ public class ServerIntegrationTest { @Test public void testPublishWithQoS1() throws IOException, InterruptedException { // startServer(); - Client client = new Client("localhost", Server.PORT); + Client client = new Client("localhost", Constants.PORT); // Client client = new Client("test.mosquitto.org", 1883); client.connect(); diff --git a/parser/pom.xml b/parser/pom.xml index 9cac38fa..2788e78d 100644 --- a/parser/pom.xml +++ b/parser/pom.xml @@ -25,11 +25,11 @@ 2.0.4 - + junit @@ -65,10 +65,9 @@ test - + 1.3 test @@ -77,7 +76,7 @@ org.fusesource.hawtbuf hawtbuf 1.9 - + diff --git a/parser/src/main/java/org/dna/mqtt/commons/Constants.java b/parser/src/main/java/org/dna/mqtt/commons/Constants.java new file mode 100644 index 00000000..64c1b898 --- /dev/null +++ b/parser/src/main/java/org/dna/mqtt/commons/Constants.java @@ -0,0 +1,9 @@ +package org.dna.mqtt.commons; + +/** + * Contains some usefull constants. + */ +public class Constants { + public static final int PORT = 1883; + public static final int DEFAULT_CONNECT_TIMEOUT = 10; +} diff --git a/parser/src/main/java/org/dna/mqtt/moquette/client/Client.java b/parser/src/main/java/org/dna/mqtt/moquette/client/Client.java index 30932908..946f32fa 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/client/Client.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/client/Client.java @@ -19,6 +19,7 @@ import org.apache.mina.filter.codec.ProtocolCodecFilter; import org.apache.mina.filter.codec.demux.DemuxingProtocolDecoder; import org.apache.mina.filter.codec.demux.DemuxingProtocolEncoder; import org.apache.mina.transport.socket.nio.NioSocketConnector; +import org.dna.mqtt.commons.Constants; import org.dna.mqtt.commons.MessageIDGenerator; import org.dna.mqtt.moquette.ConnectionException; import org.dna.mqtt.moquette.MQTTException; @@ -34,7 +35,6 @@ import org.dna.mqtt.moquette.proto.messages.PingReqMessage; import org.dna.mqtt.moquette.proto.messages.PublishMessage; import org.dna.mqtt.moquette.proto.messages.SubscribeMessage; import org.dna.mqtt.moquette.proto.messages.UnsubscribeMessage; -import org.dna.mqtt.moquette.server.Server; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -128,7 +128,7 @@ public final class Client { m_connector.setHandler(new ClientMQTTHandler(this)); m_connector.getSessionConfig().setReadBufferSize(2048); - m_connector.getSessionConfig().setIdleTime(IdleStatus.BOTH_IDLE, Server.DEFAULT_CONNECT_TIMEOUT); + m_connector.getSessionConfig().setIdleTime(IdleStatus.BOTH_IDLE, Constants.DEFAULT_CONNECT_TIMEOUT); m_scheduler = Executors.newScheduledThreadPool(NUM_SCHEDULER_TIMER_THREAD); diff --git a/pom.xml b/pom.xml index 6f6446a4..2aed6111 100644 --- a/pom.xml +++ b/pom.xml @@ -13,6 +13,7 @@ parser + broker benchmarking