Merge branch 'split_in_other_modules'

.gitignore

@@ -3,3 +3,5 @@
 /benchmarking/target/
 .idea/
 *.iml
+broker/target/
+client/target/

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 {

broker/pom.xml

@@ -0,0 +1,154 @@
+
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <relativePath>../pom.xml</relativePath>
+        <artifactId>moquette-parent</artifactId>
+        <groupId>org.dna.mqtt</groupId>
+        <version>0.1</version>
+    </parent>
+
+    <artifactId>moquette-broker</artifactId>
+    <packaging>jar</packaging>
+    <version>0.1</version>
+    <name>Moquette - broker</name>
+
+    <properties>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.dna.mqtt</groupId>
+            <artifactId>moquette-parser</artifactId>
+            <version>0.1</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.mina</groupId>
+            <artifactId>mina-core</artifactId>
+            <version>2.0.4</version>
+        </dependency>
+        
+        <dependency>
+            <groupId>com.googlecode.disruptor</groupId>
+            <artifactId>disruptor</artifactId>
+            <version>2.7.1</version>
+        </dependency>
+        
+        <!--        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-jdk14</artifactId>
+            <version>1.6.4</version>
+        </dependency>-->
+        
+        
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-log4j12</artifactId>
+            <version>1.6.4</version>
+        </dependency>
+        
+        <dependency>
+            <groupId>org.fusesource.hawtdb</groupId>
+            <artifactId>hawtdb</artifactId>
+            <version>1.6</version>
+        </dependency>
+                
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <version>4.8.2</version>
+            <scope>test</scope>
+        </dependency>
+        
+        <!-- needs extra dependencies: objenesis & hamcrest -->
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-core</artifactId>  
+            <version>1.9.0</version> 
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.fusesource.mqtt-client</groupId>
+            <artifactId>mqtt-client</artifactId>
+<!--            <version>1.4-SNAPSHOT</version>-->
+            <version>1.3</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.dna.mqtt</groupId>
+            <artifactId>moquette-client</artifactId>
+            <version>0.1</version>
+            <scope>test</scope>
+        </dependency>
+        
+        <dependency>
+            <groupId>org.fusesource.hawtbuf</groupId>
+            <artifactId>hawtbuf</artifactId>
+            <version>1.9</version>
+        </dependency>
+    </dependencies>
+    
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <configuration>
+                    <source>1.5</source>
+                    <target>1.5</target>
+                </configuration>
+            </plugin>
+            
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-surefire-plugin</artifactId>
+                <configuration>
+                    <excludes>
+                        <exclude>org/dna/mqtt/moquette/proto/TestUtils.java</exclude>
+                        <exclude>org/dna/mqtt/moquette/proto/TestUtils$*</exclude>
+                    </excludes>
+                </configuration>
+            </plugin>
+            
+            
+            <plugin>
+                <artifactId>maven-assembly-plugin</artifactId>
+                <configuration>
+                    <archive>
+                        <manifest>
+                            <mainClass>org.dna.mqtt.moquette.server.Server</mainClass>
+                        </manifest>
+                    </archive>
+                    <descriptorRefs>
+                        <descriptorRef>jar-with-dependencies</descriptorRef>
+                    </descriptorRefs>
+                </configuration>
+            </plugin>
+
+            
+        </plugins>
+    </build>
+    
+    <repositories>
+<!--        <repository>
+            <id>Fusesource Releases</id>
+            <url>http://repo.fusesource.com/nexus/content/repositories/snapshots</url>
+        </repository>-->
+<!--        <repository>
+            <id>fusesource.snapshots</id>
+            <name>FuseSource Snapshot Repository</name>
+            <url>http://repo.fusesource.com/nexus/content/repositories/snapshots</url>
+            <snapshots>
+                <enabled>true</enabled>
+            </snapshots>
+            <releases>
+                <enabled>false</enabled>
+            </releases>
+        </repository>-->
+    </repositories>
+</project>

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;
 /**
@@ -25,8 +26,8 @@ public class Server {
     
     private static final Logger LOG = LoggerFactory.getLogger(Server.class);
     
-    public static final int PORT = 1883;
-    public static final int DEFAULT_CONNECT_TIMEOUT = 10;
+//    public static final int PORT = 1883;
+//    public static final int DEFAULT_CONNECT_TIMEOUT = 10;
     public static final String STORAGE_FILE_PATH = System.getProperty("user.home") + 
             File.separator + "moquette_store.hawtdb";
     
@@ -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

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();

broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java

@@ -7,7 +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.fusesource.mqtt.client.*;
+import org.dna.mqtt.commons.Constants;
 import org.junit.After;
 import static org.junit.Assert.*;
 import org.junit.Before;
@@ -45,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();
 
@@ -68,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() {
@@ -93,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() {
@@ -106,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) {
@@ -132,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() {
@@ -145,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) {
@@ -167,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() {
@@ -189,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) {
@@ -213,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);
@@ -221,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() {
 
@@ -243,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();
 
@@ -277,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();
 
@@ -305,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
@@ -325,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();
 

client/pom.xml

@@ -0,0 +1,98 @@
+
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <relativePath>../pom.xml</relativePath>
+        <artifactId>moquette-parent</artifactId>
+        <groupId>org.dna.mqtt</groupId>
+        <version>0.1</version>
+    </parent>
+
+    <artifactId>moquette-client</artifactId>
+    <packaging>jar</packaging>
+    <version>0.1</version>
+    <name>Moquette - client</name>
+
+    <properties>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.dna.mqtt</groupId>
+            <artifactId>moquette-parser</artifactId>
+            <version>0.1</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.mina</groupId>
+            <artifactId>mina-core</artifactId>
+            <version>2.0.4</version>
+        </dependency>
+        
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-log4j12</artifactId>
+            <version>1.6.4</version>
+        </dependency>
+        
+
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <version>4.8.2</version>
+            <scope>test</scope>
+        </dependency>
+        
+        <!-- needs extra dependencies: objenesis & hamcrest -->
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-core</artifactId>  
+            <version>1.9.0</version> 
+            <scope>test</scope>
+        </dependency>
+
+    </dependencies>
+    
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <configuration>
+                    <source>1.5</source>
+                    <target>1.5</target>
+                </configuration>
+            </plugin>
+            
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-surefire-plugin</artifactId>
+                <configuration>
+                    <excludes>
+                        <exclude>org/dna/mqtt/moquette/proto/TestUtils.java</exclude>
+                        <exclude>org/dna/mqtt/moquette/proto/TestUtils$*</exclude>
+                    </excludes>
+                </configuration>
+            </plugin>
+            
+            
+            <plugin>
+                <artifactId>maven-assembly-plugin</artifactId>
+                <configuration>
+                    <archive>
+                        <manifest>
+                            <mainClass>org.dna.mqtt.moquette.client.Main</mainClass>
+                        </manifest>
+                    </archive>
+                    <descriptorRefs>
+                        <descriptorRef>jar-with-dependencies</descriptorRef>
+                    </descriptorRefs>
+                </configuration>
+            </plugin>
+
+            
+        </plugins>
+    </build>
+</project>

client/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);
         

parser/pom.xml

@@ -25,11 +25,11 @@
             <version>2.0.4</version>
         </dependency>
         
-        <dependency>
+        <!--dependency>
             <groupId>com.googlecode.disruptor</groupId>
             <artifactId>disruptor</artifactId>
             <version>2.7.1</version>
-        </dependency>
+        </dependency -->
         
         <!--        <dependency>
             <groupId>org.slf4j</groupId>
@@ -44,11 +44,11 @@
             <version>1.6.4</version>
         </dependency>
         
-        <dependency>
+        <!--dependency>
             <groupId>org.fusesource.hawtdb</groupId>
             <artifactId>hawtdb</artifactId>
             <version>1.6</version>
-        </dependency>
+        </dependency-->
                 
         <dependency>
             <groupId>junit</groupId>
@@ -65,10 +65,9 @@
             <scope>test</scope>
         </dependency>
 
-        <dependency>
+        <!--dependency>
             <groupId>org.fusesource.mqtt-client</groupId>
             <artifactId>mqtt-client</artifactId>
-<!--            <version>1.4-SNAPSHOT</version>-->
             <version>1.3</version>
             <scope>test</scope>
         </dependency>
@@ -77,7 +76,7 @@
             <groupId>org.fusesource.hawtbuf</groupId>
             <artifactId>hawtbuf</artifactId>
             <version>1.9</version>
-        </dependency>
+        </dependency-->
     </dependencies>
     
     <build>
@@ -101,23 +100,7 @@
                     </excludes>
                 </configuration>
             </plugin>
-            
-            
-            <plugin>
-                <artifactId>maven-assembly-plugin</artifactId>
-                <configuration>
-                    <archive>
-                        <manifest>
-                            <mainClass>org.dna.mqtt.moquette.server.Server</mainClass>
-                        </manifest>
-                    </archive>
-                    <descriptorRefs>
-                        <descriptorRef>jar-with-dependencies</descriptorRef>
-                    </descriptorRefs>
-                </configuration>
-            </plugin>
 
-            
         </plugins>
     </build>
     

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;
+}

pom.xml

@@ -13,6 +13,8 @@
 
     <modules>
         <module>parser</module>
+        <module>broker</module>
+        <module>client</module>
         <module>benchmarking</module>
     </modules>