diff --git a/benchmarking/src/main/java/org/dna/mqtt/bechnmark/ConsumerFuture.java b/benchmarking/src/main/java/org/dna/mqtt/bechnmark/ConsumerFuture.java
index 3447fa2a..16538949 100644
--- a/benchmarking/src/main/java/org/dna/mqtt/bechnmark/ConsumerFuture.java
+++ b/benchmarking/src/main/java/org/dna/mqtt/bechnmark/ConsumerFuture.java
@@ -4,13 +4,7 @@ import java.io.*;
 import java.net.URISyntaxException;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
-
-import org.fusesource.mqtt.client.Future;
-import org.fusesource.mqtt.client.FutureConnection;
-import org.fusesource.mqtt.client.MQTT;
-import org.fusesource.mqtt.client.Message;
-import org.fusesource.mqtt.client.QoS;
-import org.fusesource.mqtt.client.Topic;
+import org.fusesource.mqtt.client.*;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
diff --git a/parser/pom.xml b/parser/pom.xml
index 536ad89e..5a4ec1ff 100644
--- a/parser/pom.xml
+++ b/parser/pom.xml
@@ -25,6 +25,12 @@
             <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>
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 141b8d29..30932908 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
@@ -63,7 +63,7 @@ public final class Client {
     private int m_connectRetries = DEFAULT_RETRIES;
     private String m_hostname;
     private int m_port;
-    //internal mangement used for conversation with the server
+    //internal management used for conversation with the server
     private IoConnector m_connector;
     private IoSession m_session;
     private CountDownLatch m_connectBarrier;
diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java
index 6f901cc3..ce1958af 100644
--- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java
+++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java
@@ -29,7 +29,7 @@ public interface IMessaging {
     /**
      * Subscribe a client to a specified topic with a defined level
      */
-    void subscribe(String clientId, String topic, QOSType qos, boolean cleanSession);
+    void subscribe(String clientId, String topic, QOSType qos, boolean cleanSession, int messageID);
 
     /**
      * Remove all subscription to any topic the client (identified by clientID)
@@ -37,7 +37,7 @@ public interface IMessaging {
      */
     void removeSubscriptions(String clientID);
     
-    void close();
+    void stop();
 
     /**
      * Remove the clientID from topic subscription, if not previously subscribed,
diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java
index 85848203..8aec52e3 100644
--- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java
+++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java
@@ -13,10 +13,7 @@ import org.slf4j.LoggerFactory;
 
 import java.io.File;
 import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 
 import static org.dna.mqtt.moquette.messaging.spi.impl.SimpleMessaging.StoredMessage;
 
@@ -38,7 +35,8 @@ public class HawtDBStorageService implements IStorageService {
     private SortedIndex<String, PublishEvent> m_inflightStore;
 
     //persistent Map of clientID, list of Subscriptions
-    private SortedIndex<String, List<Subscription>> m_persistentSubscriptions;
+    //TODO move to Set to avoid double subscriptions
+    private SortedIndex<String, Set<Subscription>> m_persistentSubscriptions;
 
     public HawtDBStorageService() {
         String storeFile = Server.STORAGE_FILE_PATH;
@@ -84,11 +82,10 @@ public class HawtDBStorageService implements IStorageService {
     }
 
     private void initPersistentSubscriptions() {
-        BTreeIndexFactory<String, List<Subscription>> indexFactory =
-                new BTreeIndexFactory<String, List<Subscription>>();
+        BTreeIndexFactory<String, Set<Subscription>> indexFactory = new BTreeIndexFactory<String, Set<Subscription>>();
         indexFactory.setKeyCodec(StringCodec.INSTANCE);
 
-        m_persistentSubscriptions = (SortedIndex<String, List<Subscription>>) m_multiIndexFactory.openOrCreate("subscriptions", indexFactory);
+        m_persistentSubscriptions = (SortedIndex<String, Set<Subscription>>) m_multiIndexFactory.openOrCreate("subscriptions", indexFactory);
     }
 
     /**
@@ -157,12 +154,14 @@ public class HawtDBStorageService implements IStorageService {
 
     public void addNewSubscription(Subscription newSubscription, String clientID) {
         if (!m_persistentSubscriptions.containsKey(clientID)) {
-            m_persistentSubscriptions.put(clientID, new ArrayList<Subscription>());
+            m_persistentSubscriptions.put(clientID, new HashSet<Subscription>());
         }
 
-        List<Subscription> subs = m_persistentSubscriptions.get(clientID);
-        subs.add(newSubscription);
-        m_persistentSubscriptions.put(clientID, subs);
+        Set<Subscription> subs = m_persistentSubscriptions.get(clientID);
+        if (!subs.contains(newSubscription)) {
+            subs.add(newSubscription);
+            m_persistentSubscriptions.put(clientID, subs);
+        }
     }
 
     public void removeAllSubscriptions(String clientID) {
@@ -171,7 +170,7 @@ public class HawtDBStorageService implements IStorageService {
 
     public List<Subscription> retrieveAllSubscriptions() {
         List<Subscription> allSubscriptions = new ArrayList<Subscription>();
-        for (Map.Entry<String, List<Subscription>> entry : m_persistentSubscriptions) {
+        for (Map.Entry<String, Set<Subscription>> entry : m_persistentSubscriptions) {
             allSubscriptions.addAll(entry.getValue());
         }
         return allSubscriptions;
diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
index 22c92176..f3b61c64 100644
--- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
+++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
@@ -1,38 +1,37 @@
 package org.dna.mqtt.moquette.messaging.spi.impl;
 
-import org.apache.mina.core.session.IdleStatus;
-import org.apache.mina.core.session.IoSession;
-import org.dna.mqtt.moquette.messaging.spi.IMatchingCondition;
-import org.dna.mqtt.moquette.messaging.spi.IMessaging;
-import org.dna.mqtt.moquette.messaging.spi.INotifier;
-import org.dna.mqtt.moquette.messaging.spi.IStorageService;
-import org.dna.mqtt.moquette.messaging.spi.impl.SubscriptionsStore.Token;
-import org.dna.mqtt.moquette.messaging.spi.impl.events.*;
-import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
-import org.dna.mqtt.moquette.proto.messages.ConnAckMessage;
-import org.dna.mqtt.moquette.proto.messages.ConnectMessage;
-import org.dna.mqtt.moquette.proto.messages.PubAckMessage;
-import org.dna.mqtt.moquette.proto.messages.PublishMessage;
-import org.dna.mqtt.moquette.server.ConnectionDescriptor;
-import org.dna.mqtt.moquette.server.Constants;
-import org.dna.mqtt.moquette.server.IAuthenticator;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
+import com.lmax.disruptor.BatchEventProcessor;
+import com.lmax.disruptor.EventHandler;
+import com.lmax.disruptor.RingBuffer;
+import com.lmax.disruptor.SequenceBarrier;
 import java.io.Serializable;
 import java.text.ParseException;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import org.apache.mina.core.session.IdleStatus;
+import org.apache.mina.core.session.IoSession;
+import org.dna.mqtt.moquette.messaging.spi.IMatchingCondition;
+import org.dna.mqtt.moquette.messaging.spi.IMessaging;
+import org.dna.mqtt.moquette.messaging.spi.IStorageService;
+import org.dna.mqtt.moquette.messaging.spi.impl.SubscriptionsStore.Token;
+import org.dna.mqtt.moquette.messaging.spi.impl.events.*;
+import org.dna.mqtt.moquette.proto.messages.*;
+import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
+import org.dna.mqtt.moquette.server.ConnectionDescriptor;
+import org.dna.mqtt.moquette.server.Constants;
+import org.dna.mqtt.moquette.server.IAuthenticator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  *
  * @author andrea
  */
-public class SimpleMessaging implements IMessaging, Runnable {
+public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent>/*, Runnable*/ {
 
     //TODO probably move this
     public static class StoredMessage implements Serializable {
@@ -57,9 +56,9 @@ public class SimpleMessaging implements IMessaging, Runnable {
     
     private SubscriptionsStore subscriptions = new SubscriptionsStore();
     
-    private BlockingQueue<MessagingEvent> m_inboundQueue = new LinkedBlockingQueue<MessagingEvent>();
+    private RingBuffer<ValueEvent> m_ringBuffer;
 
-    private INotifier m_notifier;
+//    private INotifier m_notifier;
 
     private IStorageService m_storageService;
 
@@ -67,87 +66,86 @@ public class SimpleMessaging implements IMessaging, Runnable {
 
     private IAuthenticator m_authenticator;
     
-    public SimpleMessaging() {
-        m_storageService = new HawtDBStorageService();
-        m_storageService.initStore();
+    private ExecutorService m_executor;
+    BatchEventProcessor<ValueEvent> m_eventProcessor;
 
-        subscriptions.init(m_storageService);
+    private static SimpleMessaging INSTANCE;
+    
+    private SimpleMessaging() {
+    }
+
+    public static SimpleMessaging getInstance() {
+        if (INSTANCE == null) {
+            INSTANCE = new SimpleMessaging();
+        }
+        return INSTANCE;
+    }
+
+    public void init() {
+        m_executor = Executors.newFixedThreadPool(1);
+
+        m_ringBuffer = new RingBuffer<ValueEvent>(ValueEvent.EVENT_FACTORY, 1024 * 32);
+
+        SequenceBarrier barrier = m_ringBuffer.newBarrier();
+        m_eventProcessor = new BatchEventProcessor<ValueEvent>(m_ringBuffer, barrier, this);
+        m_ringBuffer.setGatingSequences(m_eventProcessor.getSequence());
+        m_executor.submit(m_eventProcessor);
+
+        disruptorPublish(new InitEvent());
     }
 
     
-    public void setNotifier(INotifier notifier) {
+/*    public void setNotifier(INotifier notifier) {
         m_notifier= notifier;
-    }
+    }  */
     
-    public void run() {
-        eventLoop();
+
+    private void disruptorPublish(MessagingEvent msgEvent) {
+        long sequence = m_ringBuffer.next();
+        ValueEvent event = m_ringBuffer.get(sequence);
+
+        event.setEvent(msgEvent);
+        
+        m_ringBuffer.publish(sequence); 
     }
     
 
     public void publish(String topic, byte[] message, QOSType qos, boolean retain, String clientID, IoSession session) {
-        try {
-            m_inboundQueue.put(new PublishEvent(topic, qos, message, retain, clientID, session));
-        } catch (InterruptedException ex) {
-            LOG.error(null, ex);
-        }
+        disruptorPublish(new PublishEvent(topic, qos, message, retain, clientID, session));
     }
 
     public void publish(String topic, byte[] message, QOSType qos, boolean retain, String clientID, int messageID, IoSession session) {
-        try {
-            m_inboundQueue.put(new PublishEvent(topic, qos, message, retain, clientID, messageID, session));
-        } catch (InterruptedException ex) {
-            LOG.error(null, ex);
-        }
+        disruptorPublish(new PublishEvent(topic, qos, message, retain, clientID, messageID, session));
     }
 
-    public void subscribe(String clientId, String topic, QOSType qos, boolean cleanSession) {
+    public void subscribe(String clientId, String topic, QOSType qos, boolean cleanSession, int messageID) {
         Subscription newSubscription = new Subscription(clientId, topic, qos, cleanSession);
-        try {
-            LOG.debug("subscribe invoked for topic: " + topic);
-            m_inboundQueue.put(new SubscribeEvent(newSubscription));
-        } catch (InterruptedException ex) {
-            LOG.error(null, ex);
-        }
+        LOG.debug("subscribe invoked for topic: " + topic);
+        disruptorPublish(new SubscribeEvent(newSubscription, messageID));
     }
     
     
     public void unsubscribe(String topic, String clientID) {
-        try {
-            m_inboundQueue.put(new UnsubscribeEvent(topic, clientID));
-        } catch (InterruptedException ex) {
-            LOG.error(null, ex);
-        }
+        disruptorPublish(new UnsubscribeEvent(topic, clientID));
     }
     
     
     public void disconnect(IoSession session) {
-        try {
-            m_inboundQueue.put(new DisconnectEvent(session));
-        } catch (InterruptedException ex) {
-            LOG.error(null, ex);
-        }
+        disruptorPublish(new DisconnectEvent(session));
     }
 
     //method used by hte Notifier to re-put an event on the inbound queue
     private void refill(MessagingEvent evt) throws InterruptedException {
-        m_inboundQueue.put(evt);
+        disruptorPublish(evt);
     }
 
     public void republishStored(String clientID) {
         //create the event to push
-        try {
-            m_inboundQueue.put(new RepublishEvent(clientID));
-        } catch (InterruptedException iex) {
-            LOG.error(null, iex);
-        }
+        disruptorPublish(new RepublishEvent(clientID));
     }
 
     public void connect(IoSession session, ConnectMessage msg) {
-        try {
-            m_inboundQueue.put(new ConnectEvent(session, msg));
-        } catch (InterruptedException iex) {
-            LOG.error(null, iex);
-        }
+        disruptorPublish(new ConnectEvent(session, msg));
     }
 
     /**
@@ -189,56 +187,49 @@ public class SimpleMessaging implements IMessaging, Runnable {
     }
 
     public void removeSubscriptions(String clientID) {
-        try {
-            m_inboundQueue.put(new RemoveAllSubscriptionsEvent(clientID));
-        } catch (InterruptedException ex) {
-            LOG.error(null, ex);
-        }
+        disruptorPublish(new RemoveAllSubscriptionsEvent(clientID));
     }
 
-    public void close() {
-        try {
-            m_inboundQueue.put(new CloseEvent());
-        } catch (InterruptedException ex) {
-            LOG.error(null, ex);
-        } 
+    public void stop() {
+        disruptorPublish(new StopEvent());
     }
     
-
-    protected void eventLoop() {
-        LOG.debug("Started event loop");
-        boolean interrupted = false;
-        while (!interrupted) {
-            try { 
-                MessagingEvent evt = m_inboundQueue.take();
-                if (evt instanceof PublishEvent) {
-                    processPublish((PublishEvent) evt);
-                } else if (evt instanceof SubscribeEvent) {
-                    processSubscribe((SubscribeEvent) evt);
-                } else if (evt instanceof UnsubscribeEvent) {
-                    processUnsubscribe((UnsubscribeEvent) evt);
-                } else if (evt instanceof RemoveAllSubscriptionsEvent) {
-                    processRemoveAllSubscriptions((RemoveAllSubscriptionsEvent) evt);
-                } else if (evt instanceof CloseEvent) {
-                    processClose();
-                } else if (evt instanceof DisconnectEvent) {
-                    processDisconnect((DisconnectEvent) evt);
-                } else if (evt instanceof CleanInFlightEvent) {
-                    //remove the message from inflight storage
-                    m_storageService.cleanInFlight(((CleanInFlightEvent) evt).getMsgId());
-                } else if (evt instanceof RepublishEvent) {
-                    processRepublish((RepublishEvent) evt);
-                } else if (evt instanceof ConnectEvent) {
-                    processConnect((ConnectEvent) evt);
-                }
-            } catch (InterruptedException ex) {
-                processClose();
-                interrupted = true;
-            }
+    public void onEvent(ValueEvent t, long l, boolean bln) throws Exception {
+        LOG.debug("onEvent processing event");
+        MessagingEvent evt = t.getEvent();
+        if (evt instanceof PublishEvent) {
+            processPublish((PublishEvent) evt);
+        } else if (evt instanceof SubscribeEvent) {
+            processSubscribe((SubscribeEvent) evt);
+        } else if (evt instanceof UnsubscribeEvent) {
+            processUnsubscribe((UnsubscribeEvent) evt);
+        } else if (evt instanceof RemoveAllSubscriptionsEvent) {
+            processRemoveAllSubscriptions((RemoveAllSubscriptionsEvent) evt);
+        } else if (evt instanceof StopEvent) {
+            processStop();
+        } else if (evt instanceof DisconnectEvent) {
+            processDisconnect((DisconnectEvent) evt);
+        } else if (evt instanceof CleanInFlightEvent) {
+            //remove the message from inflight storage
+            m_storageService.cleanInFlight(((CleanInFlightEvent) evt).getMsgId());
+        } else if (evt instanceof RepublishEvent) {
+            processRepublish((RepublishEvent) evt);
+        } else if (evt instanceof ConnectEvent) {
+            processConnect((ConnectEvent) evt);
+        } else if (evt instanceof InitEvent) {
+            processInit();
         }
     }
 
-    private void processConnect(ConnectEvent evt) {
+    private void processInit() {
+        m_storageService = new HawtDBStorageService();
+        m_storageService.initStore();
+
+        subscriptions.init(m_storageService);
+    }
+
+
+    protected void processConnect(ConnectEvent evt) {
         ConnectMessage msg = evt.getMessage();
         IoSession session = evt.getSession();
 
@@ -368,16 +359,29 @@ public class SimpleMessaging implements IMessaging, Runnable {
         //scans retained messages to be published to the new subscription
         Collection<StoredMessage> messages = m_storageService.searchMatching(new IMatchingCondition() {
             public boolean match(String key) {
-                return matchTopics(key, topic);  //To change body of implemented methods use File | Settings | File Templates.
+                return matchTopics(key, topic);
             }
         });
 
         for (StoredMessage storedMsg : messages) {
             //fire the as retained the message
             LOG.debug("Inserting NotifyEvent into outbound for topic " + topic);
-            m_notifier.notify(new NotifyEvent(newSubscription.clientId, topic, storedMsg.getQos(),
+            notify(new NotifyEvent(newSubscription.clientId, topic, storedMsg.getQos(),
                     storedMsg.getPayload(), true));
         }
+
+//        //ack the client
+//        SubAckMessage ackMessage = new SubAckMessage();
+//        ackMessage.setMessageID(evt.getMessageID());
+//
+//        //TODO by now it handles only QoS 0 messages
+//        /*for (int i = 0; i < evt.subscriptions().size(); i++) {
+//            ackMessage.addType(QOSType.MOST_ONE);
+//        } */
+//        LOG.info("replying with SubAct to MSG ID {0}", evt.getMessageID());
+//        //session.write(ackMessage);
+//        IoSession session = m_clientIDs.get(evt.getSubscription().clientId).getSession();
+//        session.write(ackMessage).awaitUninterruptibly();
     }
     
     protected void processUnsubscribe(UnsubscribeEvent evt) {
@@ -405,9 +409,12 @@ public class SimpleMessaging implements IMessaging, Runnable {
         subscriptions.disconnect(clientID);
     }
     
-    private void processClose() {
-        LOG.debug("processClose invoked");
+    private void processStop() {
+        LOG.debug("processStop invoked");
         m_storageService.close();
+
+//        m_eventProcessor.halt();
+        m_executor.shutdown();
     }
 
     private void processRepublish(RepublishEvent evt) throws InterruptedException {
@@ -417,7 +424,7 @@ public class SimpleMessaging implements IMessaging, Runnable {
         }
 
         for (PublishEvent pubEvt : publishedEvents) {
-            m_notifier.notify(new NotifyEvent(pubEvt.getClientID(), pubEvt.getTopic(), pubEvt.getQos(),
+            notify(new NotifyEvent(pubEvt.getClientID(), pubEvt.getTopic(), pubEvt.getQos(),
                     pubEvt.getMessage(), false, pubEvt.getMessageID()));
         }
     }
diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ValueEvent.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ValueEvent.java
new file mode 100644
index 00000000..1ec5d6cb
--- /dev/null
+++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ValueEvent.java
@@ -0,0 +1,29 @@
+package org.dna.mqtt.moquette.messaging.spi.impl;
+
+import com.lmax.disruptor.EventFactory;
+import org.dna.mqtt.moquette.messaging.spi.impl.events.MessagingEvent;
+
+/**
+ * Carrier value object for the RingBuffer.
+ *
+ * @author andrea
+ */
+public final class ValueEvent {
+
+    private MessagingEvent m_event;
+
+    public MessagingEvent getEvent() {
+        return m_event;
+    }
+
+    public void setEvent(MessagingEvent event) {
+        m_event = event;
+    }
+    
+    public final static EventFactory<ValueEvent> EVENT_FACTORY = new EventFactory<ValueEvent>() {
+
+        public ValueEvent newInstance() {
+            return new ValueEvent();
+        }
+    };
+}
diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/InitEvent.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/InitEvent.java
new file mode 100644
index 00000000..94d7ce63
--- /dev/null
+++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/InitEvent.java
@@ -0,0 +1,6 @@
+package org.dna.mqtt.moquette.messaging.spi.impl.events;
+
+/**
+ */
+public class InitEvent extends MessagingEvent {
+}
diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/CloseEvent.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/StopEvent.java
similarity index 51%
rename from parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/CloseEvent.java
rename to parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/StopEvent.java
index 8ca2119e..d422abc5 100644
--- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/CloseEvent.java
+++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/StopEvent.java
@@ -1,9 +1,10 @@
 package org.dna.mqtt.moquette.messaging.spi.impl.events;
 
 /**
+ * Poison pill to shutdown the event loop
  *
  * @author andrea
  */
-public class CloseEvent extends MessagingEvent {
+public class StopEvent extends MessagingEvent {
     
 }
diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/SubscribeEvent.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/SubscribeEvent.java
index 334e7e48..fba93dd8 100644
--- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/SubscribeEvent.java
+++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/SubscribeEvent.java
@@ -9,13 +9,19 @@ import org.dna.mqtt.moquette.messaging.spi.impl.Subscription;
 public class SubscribeEvent extends MessagingEvent {
 
     Subscription m_subscription;
+
+    int m_messageID;
     
-    public SubscribeEvent(Subscription subscription) {
+    public SubscribeEvent(Subscription subscription, int messageID) {
         m_subscription = subscription;
+        m_messageID = messageID;
     }
 
     public Subscription getSubscription() {
         return m_subscription;
     }
-    
+
+    public int getMessageID() {
+        return m_messageID;
+    }
 }
diff --git a/parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java b/parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java
index b86b90e5..d97efed9 100644
--- a/parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java
+++ b/parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java
@@ -75,7 +75,7 @@ public class MQTTHandler extends IoHandlerAdapter/* implements INotifier*/ {
             String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
             boolean cleanSession = (Boolean) session.getAttribute(Constants.CLEAN_SESSION);
             m_messaging.subscribe(clientID, req.getTopic(), AbstractMessage.QOSType.values()[req.getQos()],
-                    cleanSession);
+                    cleanSession, msg.getMessageID());
         }
 
         //ack the client
diff --git a/parser/src/main/java/org/dna/mqtt/moquette/server/Server.java b/parser/src/main/java/org/dna/mqtt/moquette/server/Server.java
index e7c61039..81184f64 100644
--- a/parser/src/main/java/org/dna/mqtt/moquette/server/Server.java
+++ b/parser/src/main/java/org/dna/mqtt/moquette/server/Server.java
@@ -3,10 +3,9 @@ package org.dna.mqtt.moquette.server;
 import java.io.File;
 import java.io.IOException;
 import java.net.InetSocketAddress;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
+
 import org.apache.mina.core.service.IoAcceptor;
+import org.apache.mina.core.service.IoServiceStatistics;
 import org.apache.mina.core.session.IdleStatus;
 import org.apache.mina.core.session.IoSession;
 import org.apache.mina.filter.codec.ProtocolCodecFilter;
@@ -14,7 +13,6 @@ import org.apache.mina.filter.codec.demux.DemuxingProtocolDecoder;
 import org.apache.mina.filter.codec.demux.DemuxingProtocolEncoder;
 import org.apache.mina.transport.socket.nio.NioSocketAcceptor;
 import org.dna.mqtt.moquette.messaging.spi.impl.SimpleMessaging;
-import org.dna.mqtt.moquette.messaging.spi.impl.events.MessagingEvent;
 import org.dna.mqtt.moquette.proto.*;
 import org.dna.mqtt.moquette.proto.messages.*;
 import org.slf4j.Logger;
@@ -36,7 +34,7 @@ public class Server {
     //private static final int NOTIFIER_POOL_SIZE = 1;
     private IoAcceptor m_acceptor;
     SimpleMessaging messaging;
-    Thread messagingEventLoop;
+//    Thread messagingEventLoop;
     //private ExecutorService m_notifierPool/* = Executors.newFixedThreadPool(NOTIFIER_POOL_SIZE)*/;
     
     public static void main(String[] args) throws IOException {
@@ -68,19 +66,22 @@ public class Server {
         m_acceptor.getFilterChain().addLast( "codec", new ProtocolCodecFilter(encoder, decoder));
 
         MQTTHandler handler = new MQTTHandler();
-        messaging = new SimpleMessaging();
+        messaging = SimpleMessaging.getInstance();
+        messaging.init();
         //TODO fix this hugly wiring
         handler.setMessaging(messaging);
 //        messaging.setNotifier(handler);
-        messagingEventLoop = new Thread(messaging);
-        messagingEventLoop.setName("Event Loop" + System.currentTimeMillis());
-        messagingEventLoop.start();
+//        messagingEventLoop = new Thread(messaging);
+//        messagingEventLoop.setName("Event Loop" + System.currentTimeMillis());
+//        messagingEventLoop.start();
         
         m_acceptor.setHandler(handler);
         ((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.getStatistics().setThroughputCalculationInterval(10);
+        m_acceptor.getStatistics().updateThroughput(System.currentTimeMillis());
         m_acceptor.bind( new InetSocketAddress(PORT) );
         LOG.info("Server binded");
         
@@ -96,23 +97,30 @@ public class Server {
     protected void stopServer() {
         LOG.info("Server stopping...");
         
-        messaging.close();
-        messagingEventLoop.interrupt();
-        LOG.info("shutting down evet loop");
-        try {
-            messagingEventLoop.join();
-        } catch (InterruptedException ex) {
-            LOG.error(null, ex);
-        }
+        messaging.stop();
+        //messaging.stop();
+//        messagingEventLoop.interrupt();
+//        LOG.info("shutting down evet loop");
+//        try {
+//            messagingEventLoop.join();
+//        } catch (InterruptedException ex) {
+//            LOG.error(null, ex);
+//        }
         
         /*m_notifierPool.shutdown();*/
         
+        //log statistics
+        IoServiceStatistics statistics  = m_acceptor.getStatistics();
+        statistics.updateThroughput(System.currentTimeMillis());
+        System.out.println(String.format("Total read bytes: %d, read throughtput: %f (b/s)", statistics.getReadBytes(), statistics.getReadBytesThroughput()));
+        System.out.println(String.format("Total read msgs: %d, read msg throughtput: %f (msg/s)", statistics.getReadMessages(), statistics.getReadMessagesThroughput()));
+        
         for(IoSession session: m_acceptor.getManagedSessions().values()) {
             if(session.isConnected() && !session.isClosing()){
                 session.close(false);
             }
         }
-        
+
         m_acceptor.unbind();
         m_acceptor.dispose();
         LOG.info("Server stopped");
diff --git a/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java b/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java
index 924427c7..c7d834f5 100644
--- a/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java
+++ b/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java
@@ -1,18 +1,24 @@
 package org.dna.mqtt.moquette.messaging.spi.impl;
 
-import java.util.concurrent.BlockingQueue;
-
-import org.dna.mqtt.moquette.messaging.spi.INotifier;
-import org.dna.mqtt.moquette.messaging.spi.impl.events.MessagingEvent;
-import org.dna.mqtt.moquette.messaging.spi.impl.events.NotifyEvent;
-import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent;
-import org.dna.mqtt.moquette.messaging.spi.impl.events.SubscribeEvent;
+import org.apache.mina.core.filterchain.IoFilterAdapter;
+import org.apache.mina.core.session.DummySession;
+import org.apache.mina.core.session.IoSession;
+import org.apache.mina.core.write.WriteRequest;
+import org.dna.mqtt.moquette.messaging.spi.impl.events.*;
+import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
 import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
 import static org.junit.Assert.*;
+
+import org.dna.mqtt.moquette.proto.messages.ConnAckMessage;
+import org.dna.mqtt.moquette.proto.messages.ConnectMessage;
+import org.dna.mqtt.moquette.proto.messages.SubscribeMessage;
+import org.dna.mqtt.moquette.server.ConnectionDescriptor;
+import org.dna.mqtt.moquette.server.Constants;
+import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
-import org.mockito.ArgumentCaptor;
-import static org.mockito.Mockito.*;
+
+import static org.junit.Assert.assertEquals;
 
 /**
  *
@@ -24,15 +30,53 @@ public class SimpleMessagingTest {
     final static String FAKE_TOPIC = "/news";
     SimpleMessaging messaging;
 
+    byte m_returnCode;
+    IoSession m_session;
+//    MQTTHandler m_handler;
+    ConnectMessage connMsg;
+
+    AbstractMessage m_receivedMessage;
+
     @Before
-    public void setUp() {
-        messaging = new SimpleMessaging();
+    public void setUp() throws InterruptedException {
+        messaging = SimpleMessaging.getInstance();
+        messaging.init();
+        connMsg = new ConnectMessage();
+        connMsg.setProcotolVersion((byte) 0x03);
+
+        m_session = new DummySession();
+
+        m_session.getFilterChain().addFirst("MessageCatcher", new IoFilterAdapter() {
+
+            @Override
+            public void filterWrite(NextFilter nextFilter, IoSession session,
+                                    WriteRequest writeRequest) throws Exception {
+                try {
+                    m_receivedMessage = (AbstractMessage) writeRequest.getMessage();
+                    if (m_receivedMessage instanceof ConnAckMessage) {
+                        ConnAckMessage buf = (ConnAckMessage) m_receivedMessage;
+                        m_returnCode = buf.getReturnCode();
+                    }
+                } catch (Exception ex) {
+                    throw new AssertionError("Wrong return code");
+                }
+            }
+        });
+
+        //sleep to let the messaging batch processor to process the initEvent
+        Thread.sleep(300);
+    }
+
+    @After
+    public void tearDown() {
+        m_receivedMessage = null;
+        messaging.stop();
     }
 
     @Test
     public void testSubscribe() {
         //Exercise
-        SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false));
+        SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false), 0);
         messaging.processSubscribe(evt);
 
         //Verify
@@ -42,7 +86,7 @@ public class SimpleMessagingTest {
 
     @Test
     public void testDoubleSubscribe() {
-        SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false));
+        SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false), 0);
         messaging.processSubscribe(evt);
 
         //Exercise
@@ -55,26 +99,28 @@ public class SimpleMessagingTest {
 
     @Test
     public void testPublish() throws InterruptedException {
-        SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false));
+        //simulate a connect that register a clientID to an IoSession
+        ConnectionDescriptor connDescr = new ConnectionDescriptor(FAKE_CLIENT_ID, m_session, true);
+        messaging.m_clientIDs.put(FAKE_CLIENT_ID, connDescr);
+
+        SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false), 0);
         messaging.processSubscribe(evt);
-        INotifier notifier = mock(INotifier.class);
-        messaging.setNotifier(notifier);
 
         //Exercise
         PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, QOSType.MOST_ONE, "Hello".getBytes(), false, "FakeCLI", null);
         messaging.processPublish(pubEvt);
 
         //Verify
-        ArgumentCaptor<NotifyEvent> argument = ArgumentCaptor.forClass(NotifyEvent.class);
-        verify(notifier).notify(argument.capture());
+        assertNotNull(m_receivedMessage);
+        //TODO check received message attributes
     }
-    
+
     @Test
     public void testMatchTopics_simple() {
         assertTrue(messaging.matchTopics("/", "/"));
         assertTrue(messaging.matchTopics("/finance", "/finance"));
     }
-    
+
     @Test
     public void testMatchTopics_multi() {
         assertTrue(messaging.matchTopics("finance", "#"));
@@ -82,8 +128,8 @@ public class SimpleMessagingTest {
         assertTrue(messaging.matchTopics("finance/stock", "finance/#"));
         assertTrue(messaging.matchTopics("finance/stock/ibm", "finance/#"));
     }
-    
-    
+
+
     @Test
     public void testMatchTopics_single() {
         assertTrue(messaging.matchTopics("finance", "+"));
@@ -94,4 +140,74 @@ public class SimpleMessagingTest {
         assertTrue(messaging.matchTopics("/finance/stock/ibm", "/finance/+/ibm"));
         assertFalse(messaging.matchTopics("/finance/stock", "+"));
     }
+
+    @Test
+    public void testHandleConnect_BadProtocol() {
+        connMsg.setProcotolVersion((byte) 0x02);
+
+        //Exercise
+        ConnectEvent pubEvt = new ConnectEvent(m_session, connMsg);
+        messaging.processConnect(pubEvt);
+
+        //Verify
+        assertEquals(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION, m_returnCode);
+    }
+
+    @Test
+    public void testConnect_badClientID() {
+        connMsg.setClientID("extremely_long_clientID_greater_than_23");
+
+        //Exercise
+        ConnectEvent pubEvt = new ConnectEvent(m_session, connMsg);
+        messaging.processConnect(pubEvt);
+
+        //Verify
+        assertEquals(ConnAckMessage.IDENTIFIER_REJECTED, m_returnCode);
+    }
+
+    @Test
+    public void testWill() {
+        connMsg.setClientID("123");
+        connMsg.setWillFlag(true);
+        connMsg.setWillTopic("topic");
+        connMsg.setWillMessage("Topic message");
+        //IMessaging mockedMessaging = mock(IMessaging.class);
+
+        //Exercise
+        //m_handler.setMessaging(mockedMessaging);
+        ConnectEvent pubEvt = new ConnectEvent(m_session, connMsg);
+        messaging.processConnect(pubEvt);
+
+        //Verify
+        assertEquals(ConnAckMessage.CONNECTION_ACCEPTED, m_returnCode);
+        //TODO verify the call
+        /*verify(mockedMessaging).publish(eq("topic"), eq("Topic message".getBytes()),
+                any(AbstractMessage.QOSType.class), anyBoolean(), eq("123"), any(IoSession.class));*/
+    }
+
+
+    @Test
+    public void testHandleSubscribe() {
+        String topicName = "/news";
+        SubscribeMessage subscribeMsg = new SubscribeMessage();
+        subscribeMsg.setMessageID(0x555);
+        subscribeMsg.addSubscription(new SubscribeMessage.Couple(
+                (byte)AbstractMessage.QOSType.EXACTLY_ONCE.ordinal(), topicName));
+        //IMessaging mockedMessaging = mock(IMessaging.class);
+
+        m_session.setAttribute(Constants.ATTR_CLIENTID, "fakeID");
+        m_session.setAttribute(Constants.CLEAN_SESSION, false);
+        Subscription subscription = new Subscription("fakeID", "/news", QOSType.EXACTLY_ONCE, true);
+
+        //Exercise
+        //m_handler.setMessaging(mockedMessaging);
+        SubscribeEvent pubEvt = new SubscribeEvent(subscription, 0);
+        messaging.processSubscribe(pubEvt);
+
+        //Verify
+        /*ArgumentCaptor<AbstractMessage.QOSType> argument = ArgumentCaptor.forClass(AbstractMessage.QOSType.class);
+        verify(mockedMessaging).subscribe(eq("fakeID"), eq(topicName), argument.capture(), eq(false));
+        assertEquals(AbstractMessage.QOSType.EXACTLY_ONCE, argument.getValue());*/
+        //TODO verify the message ACK and corresponding QoS!!
+    }
 }
diff --git a/parser/src/test/java/org/dna/mqtt/moquette/server/MQTTHandlerTest.java b/parser/src/test/java/org/dna/mqtt/moquette/server/MQTTHandlerTest.java
deleted file mode 100644
index dd227bf3..00000000
--- a/parser/src/test/java/org/dna/mqtt/moquette/server/MQTTHandlerTest.java
+++ /dev/null
@@ -1,120 +0,0 @@
-package org.dna.mqtt.moquette.server;
-
-import org.mockito.ArgumentCaptor;
-import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
-import org.apache.mina.core.write.WriteRequest;
-import org.dna.mqtt.moquette.proto.messages.ConnAckMessage;
-import org.apache.mina.core.filterchain.IoFilterAdapter;
-import org.apache.mina.core.session.DummySession;
-import org.apache.mina.core.session.IoSession;
-import org.dna.mqtt.moquette.messaging.spi.IMessaging;
-import org.dna.mqtt.moquette.proto.messages.ConnectMessage;
-import org.dna.mqtt.moquette.proto.messages.SubscribeMessage;
-import org.junit.Before;
-import org.junit.Test;
-import static org.junit.Assert.*;
-import static org.mockito.Mockito.*;
-
-/**
- *
- * @author andrea
- */
-public class MQTTHandlerTest {
-
-    byte m_returnCode;
-    IoSession m_session;
-    MQTTHandler m_handler;
-    ConnectMessage connMsg;
-    
-    @Before
-    public void setUp() {
-        m_handler = new MQTTHandler();
-        connMsg = new ConnectMessage();
-        connMsg.setProcotolVersion((byte) 0x03);
-        
-        m_session = new DummySession();
-        m_session.getFilterChain().addFirst("MessageCatcher", new IoFilterAdapter() {
-
-            @Override
-            public void filterWrite(NextFilter nextFilter, IoSession session,
-                    WriteRequest writeRequest) throws Exception {
-                try {
-                    ConnAckMessage buf = (ConnAckMessage) writeRequest.getMessage();
-                    m_returnCode = buf.getReturnCode();
-                } catch (Exception ex) {
-                    throw new AssertionError("Wrong return code");
-                }
-            }
-        });
-    }
-    
-    @Test
-    public void testHandleConnect_BadProtocol() {
-        connMsg.setProcotolVersion((byte) 0x02);
-
-        //Exercise
-        m_handler.handleConnect(m_session, connMsg);
-        
-        //Verify
-        assertEquals(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION, m_returnCode);
-    }
-    
-    
-    @Test
-    public void testConnect_badClientID() {
-        connMsg.setClientID("extremely_long_clientID_greater_than_23");
-
-        //Exercise
-        m_handler.handleConnect(m_session, connMsg);
-        
-        //Verify
-        assertEquals(ConnAckMessage.IDENTIFIER_REJECTED, m_returnCode);
-    }
-    
-    @Test
-    public void testWill() {
-        connMsg.setClientID("123");
-        connMsg.setWillFlag(true);
-        connMsg.setWillTopic("topic");
-        connMsg.setWillMessage("Topic message");
-        IMessaging mockedMessaging = mock(IMessaging.class);
-
-        //Exercise
-        m_handler.setMessaging(mockedMessaging);
-        m_handler.handleConnect(m_session, connMsg);
-        
-        //Verify
-        assertEquals(ConnAckMessage.CONNECTION_ACCEPTED, m_returnCode);
-        verify(mockedMessaging).publish(eq("topic"), eq("Topic message".getBytes()), 
-                any(AbstractMessage.QOSType.class), anyBoolean(), eq("123"), any(IoSession.class));
-    }
-    
-    
-    @Test
-    public void testHandleSubscribe() {
-        String topicName = "/news";
-        SubscribeMessage subscribeMsg = new SubscribeMessage();
-        subscribeMsg.setMessageID(0x555);
-        subscribeMsg.addSubscription(new SubscribeMessage.Couple(
-                (byte)AbstractMessage.QOSType.EXACTLY_ONCE.ordinal(), topicName));
-        IMessaging mockedMessaging = mock(IMessaging.class);
-        
-        m_session.setAttribute(Constants.ATTR_CLIENTID, "fakeID");
-        m_session.setAttribute(Constants.CLEAN_SESSION, false);
-
-        //Exercise
-        m_handler.setMessaging(mockedMessaging);
-        m_handler.handleSubscribe(m_session, subscribeMsg);
-        
-        //Verify
-        ArgumentCaptor<AbstractMessage.QOSType> argument = ArgumentCaptor.forClass(AbstractMessage.QOSType.class);
-        verify(mockedMessaging).subscribe(eq("fakeID"), eq(topicName), argument.capture(), eq(false));
-        assertEquals(AbstractMessage.QOSType.EXACTLY_ONCE, argument.getValue());
-        //TODO verify the message ACK and corresponding QoS!!
-    }
-    
-//    @Test
-//    public void testHandleSubscribeMultipleTopics() {
-//        fail("TOBE implemented");
-//    }
-}
diff --git a/parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java b/parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java
index 28aaa6df..966507ab 100644
--- a/parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java
+++ b/parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java
@@ -48,29 +48,29 @@ public class ServerIntegrationTest {
         Client client = new Client("localhost", Server.PORT);
 //        Client client = new Client("test.mosquitto.org", 1883);
         client.connect();
-        
-        
+
+
         client.subscribe("/topic", new IPublishCallback() {
 
             public void published(String topic, byte[] message) {
                 received = true;
             }
         });
-        
+
         client.publish("/topic", "Test my payload".getBytes());
-        
+
         client.close();
         client.shutdown();
-        
+
         assertTrue(received);
     }
-    
-    
+
+
     @Test
     public void testCleanSession_maintainClientSubscriptions() {
         Client client = new Client("localhost", Server.PORT/*, "CLID_123"*/);
         client.connect(false); //without session cleanup
-        
+
         client.subscribe("/topic", new IPublishCallback() {
 
             public void published(String topic, byte[] message) {
@@ -78,24 +78,24 @@ public class ServerIntegrationTest {
             }
         });
         client.close();
-        
+
         //Exercise that the client maintain the subscriptions
-        client.connect(false); 
+        client.connect(false);
         client.publish("/topic", "Test my payload".getBytes());
         client.close();
-        
+
         //Verify
         assertTrue(received);
-        
-        //TearDown 
+
+        //TearDown
         client.shutdown();
     }
-    
+
     @Test
     public void testCleanSession_maintainClientSubscriptions_againstClientDestruction() {
         Client client = new Client("localhost", Server.PORT, "CLID_123");
         client.connect(false); //without session cleanup
-        
+
         client.subscribe("/topic", new IPublishCallback() {
 
             public void published(String topic, byte[] message) {
@@ -104,7 +104,7 @@ public class ServerIntegrationTest {
         });
         client.close();
         client.shutdown();
-        
+
         //Exercise that the client maintain the subscriptions
         client = new Client("localhost", Server.PORT, "CLID_123");
         client.register("/topic", new IPublishCallback() {
@@ -113,20 +113,20 @@ public class ServerIntegrationTest {
                 received = true;
             }
         });
-        client.connect(false); 
+        client.connect(false);
         client.publish("/topic", "Test my payload".getBytes());
         client.close();
-        
+
         //Verify
         assertTrue(received);
-        
-        //TearDown 
+
+        //TearDown
         client.shutdown();
-    } 
-    
-    
+    }
+
+
     /**
-     * Check that after a client has connected with clean session false, subscribed 
+     * Check that after a client has connected with clean session false, subscribed
      * to some topic and exited, if it reconnect with clean session true, the server
      * correctly cleanup every previous subscription
      */
@@ -134,7 +134,7 @@ public class ServerIntegrationTest {
     public void testCleanSession_correctlyClientSubscriptions() {
         Client client = new Client("localhost", Server.PORT, "CLID_123");
         client.connect(false); //without session cleanup
-        
+
         client.subscribe("/topic", new IPublishCallback() {
 
             public void published(String topic, byte[] message) {
@@ -143,7 +143,7 @@ public class ServerIntegrationTest {
         });
         client.close();
         client.shutdown();
-        
+
         //Exercise that the client maintain the subscriptions
         client = new Client("localhost", Server.PORT, "CLID_123");
         client.register("/topic", new IPublishCallback() {
@@ -152,16 +152,16 @@ public class ServerIntegrationTest {
                 received = true;
             }
         });
-        client.connect(true); 
+        client.connect(true);
         client.publish("/topic", "Test my payload".getBytes());
         client.close();
-        
+
         //Verify
         assertFalse(received);
-        
-        //TearDown 
+
+        //TearDown
         client.shutdown();
-    }    
+    }
     
     
     @Test
@@ -174,21 +174,27 @@ public class ServerIntegrationTest {
 
             public void published(String topic, byte[] message) {
                 received = true;
+                barrier.countDown();
             }
         });
+        //never called because no notification is pending
+        assertFalse(barrier.await(1, TimeUnit.SECONDS));
+
         client.close();
         client.shutdown();
         server.stopServer();
         
         server.startServer();
-        
+
+        final CountDownLatch barrier2 = new CountDownLatch(1);
+
         //Exercise that the client maintain the subscriptions
         client = new Client("localhost", Server.PORT, "CLID_123");
         client.register("/topic", new IPublishCallback() {
 
             public void published(String topic, byte[] message) {
                 received = true;
-                barrier.countDown();
+                barrier2.countDown();
             }
         });
         client.connect(false); 
@@ -196,7 +202,7 @@ public class ServerIntegrationTest {
         client.close();
         
         //Verify
-        barrier.await(1, TimeUnit.SECONDS);
+        assertTrue(barrier2.await(1, TimeUnit.SECONDS));
         assertTrue(received);
         
         //TearDown 
@@ -209,14 +215,14 @@ public class ServerIntegrationTest {
         final CountDownLatch barrier = new CountDownLatch(1);
         Client client = new Client("localhost", Server.PORT, "CLID_123");
         client.connect();
-        
+
         client.publish("/topic", "Test my payload".getBytes(), true);
         client.close();
         client.shutdown();
-        
+
         //Exercise that the client maintain the subscriptions
         client = new Client("localhost", Server.PORT, "CLID_123");
-        client.connect(); 
+        client.connect();
         client.subscribe("/topic", new IPublishCallback() {
 
             public void published(String topic, byte[] message) {
@@ -225,24 +231,24 @@ public class ServerIntegrationTest {
             }
         });
         client.close();
-        
+
         //Verify
         barrier.await(1, TimeUnit.SECONDS);
         assertTrue(received);
-        
-        //TearDown 
+
+        //TearDown
         client.shutdown();
-    } 
-    
-    
+    }
+
+
     @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("test.mosquitto.org", 1883);
         client.connect();
-        
+
         final CountDownLatch barrier = new CountDownLatch(1);
-        
+
         client.subscribe("/topic", new IPublishCallback() {
 
             public void published(String topic, byte[] message) {
@@ -250,33 +256,33 @@ public class ServerIntegrationTest {
                 barrier.countDown();
             }
         });
-        
+
         client.publish("/topic", "Test my payload".getBytes());
 
         //wait 1 second to receive the published message
         boolean unlocked = barrier.await(1000, TimeUnit.MILLISECONDS);
         assertTrue(unlocked); //we were unlocked by the message reception
         assertTrue(received);
-        
+
         //reinit the flag
         received = false;
         //unsubscrbe and republish to check no notification is raised up
         client.unsubscribe("/topic");
         client.publish("/topic", "Test my payload".getBytes());
         assertFalse(received);
-        
+
         client.close();
         client.shutdown();
     }
-    
+
     @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("test.mosquitto.org", 1883);
         client.connect();
-        
+
         final CountDownLatch barrier = new CountDownLatch(1);
-        
+
         client.subscribe("/topic", new IPublishCallback() {
 
             public void published(String topic, byte[] message) {
@@ -284,33 +290,33 @@ public class ServerIntegrationTest {
                 barrier.countDown();
             }
         });
-        
+
         client.publish("/topic", "Test my payload".getBytes());
 
         //wait 1 second to receive the published message
         boolean unlocked = barrier.await(1000, TimeUnit.MILLISECONDS);
         assertTrue(unlocked); //we were unlocked by the message reception
         assertTrue(received);
-        
+
         //unsubscrbe and republish to check no notification is raised up
         client.unsubscribe("/topic");
-        
+
         client.close();
         client.shutdown();
-        
+
         //Exercise that the client maintain the subscriptions
         client = new Client("localhost", Server.PORT, "CLID_123");
-        client.connect(); 
-        
+        client.connect();
+
         //reinit the flag
         received = false;
         final CountDownLatch barrier2 = new CountDownLatch(1);
-        
+
         client.publish("/topic", "Test my payload".getBytes());
         unlocked = barrier2.await(1000, TimeUnit.MILLISECONDS);
         assertFalse(unlocked); //we were unlocked by the timeout exipration, no message received
         assertFalse(received);
-        
+
         client.close();
         client.shutdown();
     }