diff --git a/README.md b/README.md
index deda0671..c3b752c5 100644
--- a/README.md
+++ b/README.md
@@ -38,7 +38,7 @@ To embed Moquette in another maven project is sufficient to include a repository
 <repositories>
   <repository>
     <id>bintray</id>
-    <url>http://dl.bintray.com/andsel/maven/</url>
+    <url>https://jcenter.bintray.com</url>
     <releases>
       <enabled>true</enabled>
     </releases>
@@ -55,7 +55,7 @@ Include dependency in your project:
 <dependency>
       <groupId>io.moquette</groupId>
       <artifactId>moquette-broker</artifactId>
-      <version>0.8</version>
+      <version>0.8.1</version>
 </dependency>
 ```
 
diff --git a/broker/src/main/java/io/moquette/spi/ClientSession.java b/broker/src/main/java/io/moquette/spi/ClientSession.java
index e1a15648..16c03476 100644
--- a/broker/src/main/java/io/moquette/spi/ClientSession.java
+++ b/broker/src/main/java/io/moquette/spi/ClientSession.java
@@ -16,19 +16,14 @@
 package io.moquette.spi;
 
 import io.moquette.parser.proto.messages.AbstractMessage;
-import io.moquette.parser.proto.messages.PublishMessage;
-import io.moquette.server.Constants;
+import io.moquette.spi.ISessionsStore.ClientTopicCouple;
 import io.moquette.spi.impl.subscriptions.Subscription;
 import io.moquette.spi.impl.subscriptions.SubscriptionsStore;
-import io.moquette.spi.ISessionsStore.ClientTopicCouple;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.Collection;
 import java.util.HashSet;
-import java.util.List;
 import java.util.Set;
-import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
 
 /**
@@ -61,7 +56,7 @@ public class ClientSession {
 
     private volatile boolean cleanSession;
 
-    private BlockingQueue<AbstractMessage> m_queueToPublish = new ArrayBlockingQueue<>(Constants.MAX_MESSAGE_QUEUE);
+//    private BlockingQueue<AbstractMessage> m_queueToPublish = new ArrayBlockingQueue<>(Constants.MAX_MESSAGE_QUEUE);
 
     public ClientSession(String clientID, IMessagesStore messagesStore, ISessionsStore sessionsStore,
                          boolean cleanSession) {
@@ -72,21 +67,14 @@ public class ClientSession {
     }
 
     /**
+     * Return the list of persisted publishes for the given clientID.
+     * For QoS1 and QoS2 with clean session flag, this method return the list of
+     * missed publish events while the client was disconnected.
+     *
      * @return the list of messages to be delivered for client related to the session.
      * */
-    public List<IMessagesStore.StoredMessage> storedMessages() {
-        //read all messages from enqueued store
-        Collection<MessageGUID> guids = this.m_sessionsStore.enqueued(clientID);
-        return messagesStore.listMessagesInSession(guids);
-    }
-
-    /**
-     * Remove the messages stored in a cleanSession false.
-     *
-     * @param guid the guid of the message to remove from the queue.
-     */
-    public void removeEnqueued(MessageGUID guid) {
-        this.m_sessionsStore.removeEnqueued(this.clientID, guid);
+    public BlockingQueue<IMessagesStore.StoredMessage> queue() {
+        return this.m_sessionsStore.queue(clientID);
     }
 
     @Override
@@ -173,8 +161,12 @@ public class ClientSession {
         return messagesStore.getMessageByGuid(guid);
     }
 
-    public void enqueueToDeliver(MessageGUID guid) {
-        this.m_sessionsStore.bindToDeliver(guid, this.clientID);
+    /**
+     * Enqueue a message to be sent to the client.
+     * @param message the message to enqueue.
+     * */
+    public void enqueue(IMessagesStore.StoredMessage message) {
+        this.m_sessionsStore.queue(this.clientID).add(message);
     }
 
     public IMessagesStore.StoredMessage storedMessage(int messageID) {
@@ -182,19 +174,6 @@ public class ClientSession {
         return messagesStore.getMessageByGuid(guid);
     }
 
-    /**
-     * Enqueue a message to be sent to the client.
-     * @param pubMessage the message to enqueue.
-     * @return false if the queue is full.
-     * */
-    public boolean enqueue(PublishMessage pubMessage) {
-        return m_queueToPublish.offer(pubMessage);
-    }
-
-    public AbstractMessage dequeue() {
-        return m_queueToPublish.poll();
-    }
-
     public void moveInFlightToSecondPhaseAckWaiting(int messageID) {
         m_sessionsStore.moveInFlightToSecondPhaseAckWaiting(this.clientID, messageID);
     }
diff --git a/broker/src/main/java/io/moquette/spi/IMessagesStore.java b/broker/src/main/java/io/moquette/spi/IMessagesStore.java
index 4e8ab093..d4288e07 100644
--- a/broker/src/main/java/io/moquette/spi/IMessagesStore.java
+++ b/broker/src/main/java/io/moquette/spi/IMessagesStore.java
@@ -144,15 +144,6 @@ public interface IMessagesStore {
      * */
     MessageGUID storePublishForFuture(StoredMessage storedMessage);
 
-    /**
-     * Return the list of persisted publishes for the given clientID.
-     * For QoS1 and QoS2 with clean session flag, this method return the list of 
-     * missed publish events while the client was disconnected.
-     * @param guids the list of of guid to use as search keys.
-     * @return the list of stored messages matching the passed keys.
-     */
-    List<StoredMessage> listMessagesInSession(Collection<MessageGUID> guids);
-    
     void dropMessagesInSession(String clientID);
 
     StoredMessage getMessageByGuid(MessageGUID guid);
diff --git a/broker/src/main/java/io/moquette/spi/ISessionsStore.java b/broker/src/main/java/io/moquette/spi/ISessionsStore.java
index 4e386d9b..006ab0f9 100644
--- a/broker/src/main/java/io/moquette/spi/ISessionsStore.java
+++ b/broker/src/main/java/io/moquette/spi/ISessionsStore.java
@@ -15,8 +15,8 @@
  */
 package io.moquette.spi;
 
-import java.util.Collection;
 import java.util.List;
+import java.util.concurrent.BlockingQueue;
 
 import io.moquette.spi.IMessagesStore.StoredMessage;
 import io.moquette.spi.impl.subscriptions.Subscription;
@@ -135,26 +135,14 @@ public interface ISessionsStore {
      * */
     int nextPacketID(String clientID);
 
-    /**
-     * Store the guid to be later published.
-     * @param guid the message guid to bind.
-     * @param clientID the clientID
-     * */
-    void bindToDeliver(MessageGUID guid, String clientID);
-
     /**
      * List the guids for retained messages for the session
      * @param clientID the client ID owning the queue.
-     * @return the list of enqueued message UUIDs.
+     * @return the list of queue message UUIDs.
      * */
-    Collection<MessageGUID> enqueued(String clientID);
+    BlockingQueue<StoredMessage> queue(String clientID);
 
-    /**
-     * Remove form the queue of stored messages for session.
-     * @param clientID the client ID owning the queue.
-     * @param guid the message guid to dequeue.
-     * */
-    void removeEnqueued(String clientID, MessageGUID guid);
+    void dropQueue(String clientID);
 
     void moveInFlightToSecondPhaseAckWaiting(String clientID, int messageID);
 
diff --git a/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java b/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java
index ec3bbac0..10db65f7 100644
--- a/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java
+++ b/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java
@@ -9,8 +9,11 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.nio.ByteBuffer;
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
+import java.util.Queue;
+import java.util.concurrent.BlockingQueue;
 
 class InternalRepublisher {
 
@@ -39,8 +42,11 @@ class InternalRepublisher {
         }
     }
 
-    void publishStored(ClientSession clientSession, List<IMessagesStore.StoredMessage> publishedEvents) {
-        for (IMessagesStore.StoredMessage pubEvt : publishedEvents) {
+    void publishStored(ClientSession clientSession, BlockingQueue<IMessagesStore.StoredMessage> publishedEvents) {
+        List<IMessagesStore.StoredMessage> storedPublishes = new ArrayList<>();
+        publishedEvents.drainTo(storedPublishes);
+
+        for (IMessagesStore.StoredMessage pubEvt : storedPublishes) {
             //put in flight zone
             LOG.trace("Adding to inflight <{}>", pubEvt.getMessageID());
             clientSession.inFlightAckWaiting(pubEvt.getGuid(), pubEvt.getMessageID());
@@ -61,7 +67,7 @@ class InternalRepublisher {
         return createPublishForQos(storedMessage.getTopic(), storedMessage.getQos(), storedMessage.getMessage(), true);
     }
 
-    private PublishMessage createPublishForQos(String topic, AbstractMessage.QOSType qos, ByteBuffer message, boolean retained) {
+    public static PublishMessage createPublishForQos(String topic, AbstractMessage.QOSType qos, ByteBuffer message, boolean retained) {
         PublishMessage pubMessage = new PublishMessage();
         pubMessage.setRetainFlag(retained);
         pubMessage.setTopicName(topic);
diff --git a/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java b/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java
index 1c4146a4..5f000d0d 100644
--- a/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java
+++ b/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java
@@ -48,6 +48,7 @@ class MessagesPublisher {
         final ByteBuffer origMessage = pubMsg.getMessage();
 
         //if QoS 1 or 2 store the message
+        //TODO perhaps this block is not needed.
         MessageGUID guid = null;
         if (publishingQos != AbstractMessage.QOSType.MOST_ONE) {
             guid = m_messagesStore.storePublishForFuture(pubMsg);
@@ -76,7 +77,7 @@ class MessagesPublisher {
             } else {
                 if (!targetSession.isCleanSession()) {
                     //store the message in targetSession queue to deliver
-                    targetSession.enqueueToDeliver(guid);
+                    targetSession.enqueue(pubMsg);
                 }
             }
         }
diff --git a/broker/src/main/java/io/moquette/spi/impl/PersistentQueueMessageSender.java b/broker/src/main/java/io/moquette/spi/impl/PersistentQueueMessageSender.java
index c674bd98..30f5bae2 100644
--- a/broker/src/main/java/io/moquette/spi/impl/PersistentQueueMessageSender.java
+++ b/broker/src/main/java/io/moquette/spi/impl/PersistentQueueMessageSender.java
@@ -10,6 +10,7 @@ import org.slf4j.LoggerFactory;
 import java.util.concurrent.ConcurrentMap;
 
 import static io.moquette.parser.proto.messages.AbstractMessage.QOSType.MOST_ONE;
+import static io.moquette.spi.impl.ProtocolProcessor.asStoredMessage;
 
 class PersistentQueueMessageSender {
 
@@ -48,7 +49,7 @@ class PersistentQueueMessageSender {
         } else if (pubMessage.getQos() != MOST_ONE) {
             //enqueue to the client session
             LOG.debug("enqueue to client session");
-            clientsession.enqueue(pubMessage);
+            clientsession.enqueue(asStoredMessage(pubMessage));
         }
     }
 }
diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
index a7e91a0b..e09eb0ca 100644
--- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
@@ -17,9 +17,7 @@ package io.moquette.spi.impl;
 
 import java.nio.ByteBuffer;
 import java.util.*;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.*;
 
 import io.moquette.interception.InterceptHandler;
 import io.moquette.parser.proto.messages.*;
@@ -36,6 +34,8 @@ import io.moquette.spi.impl.subscriptions.Subscription;
 
 import static io.moquette.parser.netty.Utils.VERSION_3_1;
 import static io.moquette.parser.netty.Utils.VERSION_3_1_1;
+import static io.moquette.spi.impl.InternalRepublisher.createPublishForQos;
+
 import io.moquette.interception.messages.InterceptAcknowledgedMessage;
 import io.moquette.parser.proto.messages.AbstractMessage.QOSType;
 import io.netty.channel.Channel;
@@ -43,7 +43,6 @@ import io.netty.channel.ChannelPipeline;
 import io.netty.handler.timeout.IdleStateHandler;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import sun.misc.MessageUtils;
 
 /**
  * Class responsible to handle the logic of MQTT protocol it's the director of
@@ -406,7 +405,7 @@ public class ProtocolProcessor {
      * */
     private void republishStoredInSession(ClientSession clientSession) {
         LOG.trace("republishStoredInSession for client <{}>", clientSession);
-        List<IMessagesStore.StoredMessage> publishedEvents = clientSession.storedMessages();
+        BlockingQueue<StoredMessage> publishedEvents = clientSession.queue();
         if (publishedEvents.isEmpty()) {
             LOG.info("No stored messages for client <{}>", clientSession.clientID);
             return;
@@ -414,9 +413,6 @@ public class ProtocolProcessor {
 
         LOG.info("republishing stored messages to client <{}>", clientSession.clientID);
         this.internalRepublisher.publishStored(clientSession, publishedEvents);
-        for (IMessagesStore.StoredMessage pubEvt : publishedEvents) {
-            clientSession.removeEnqueued(pubEvt.getGuid());
-        }
     }
 
     public void processPubAck(Channel channel, PubAckMessage msg) {
@@ -438,7 +434,7 @@ public class ProtocolProcessor {
         m_interceptor.notifyMessageAcknowledged(new InterceptAcknowledgedMessage(inflightMsg, topic, username));
     }
 
-    static IMessagesStore.StoredMessage asStoredMessage(PublishMessage msg) {
+    public static IMessagesStore.StoredMessage asStoredMessage(PublishMessage msg) {
         IMessagesStore.StoredMessage stored = new IMessagesStore.StoredMessage(msg.getPayload().array(), msg.getQos(), msg.getTopicName());
         stored.setRetained(msg.isRetainFlag());
         stored.setMessageID(msg.getMessageID());
@@ -641,10 +637,7 @@ public class ProtocolProcessor {
 
         if (descriptor.cleanSession) {
             LOG.debug("Removing messages in session's queue for client <{}>", clientID);
-            Collection<MessageGUID> queue = this.m_sessionsStore.enqueued(clientID);
-            for (MessageGUID guid : queue) {
-                this.m_sessionsStore.removeEnqueued(clientID, guid);
-            }
+            this.m_sessionsStore.dropQueue(clientID);
             LOG.debug("Removed messages in session for client's queue <{}>", clientID);
         }
         return true;
@@ -848,11 +841,14 @@ public class ProtocolProcessor {
         ClientSession clientSession = m_sessionsStore.sessionForClient(clientID);
         boolean emptyQueue = false;
         while (channel.isWritable()  && !emptyQueue) {
-            AbstractMessage msg = clientSession.dequeue();
+            StoredMessage msg = clientSession.queue().poll();
             if (msg == null) {
                 emptyQueue = true;
             } else {
-                channel.write(msg);
+                //recreate a publish from stored publish in queue
+                boolean retained = m_messagesStore.getMessageByGuid(msg.getGuid()) != null;
+                PublishMessage pubMsg = createPublishForQos(msg.getTopic(), msg.getQos(), msg.getMessage(), retained);
+                channel.write(pubMsg);
             }
         }
         channel.flush();
diff --git a/broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java b/broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java
index c035a418..3991685a 100644
--- a/broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java
+++ b/broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java
@@ -78,7 +78,7 @@ class MapDBMessagesStore implements IMessagesStore {
         LOG.debug("storePublishForFuture store evt {}", storedMessage);
         if (storedMessage.getClientID() == null) {
             LOG.error("persisting a message without a clientID, bad programming error msg: {}", storedMessage);
-            throw new IllegalArgumentException("\"persisting a message without a clientID, bad programming error");
+            throw new IllegalArgumentException("persisting a message without a clientID, bad programming error");
         }
         MessageGUID guid = new MessageGUID(UUID.randomUUID().toString());
         storedMessage.setGuid(guid);
@@ -89,15 +89,6 @@ class MapDBMessagesStore implements IMessagesStore {
         return guid;
     }
 
-    @Override
-    public List<StoredMessage> listMessagesInSession(Collection<MessageGUID> guids) {
-        List<StoredMessage> ret = new ArrayList<>();
-        for (MessageGUID guid : guids) {
-            ret.add(m_persistentMessageStore.get(guid));
-        }
-        return ret;
-    }
-
     @Override
     public void dropMessagesInSession(String clientID) {
         ConcurrentMap<Integer, MessageGUID> messageIdToGuid = m_db.getHashMap(MapDBSessionsStore.messageId2GuidsMapName(clientID));
diff --git a/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java b/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java
index 15dbbde5..9a6ca0a8 100644
--- a/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java
+++ b/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java
@@ -16,6 +16,7 @@
 package io.moquette.spi.persistence;
 
 import io.moquette.spi.ClientSession;
+import io.moquette.spi.IMessagesStore;
 import io.moquette.spi.IMessagesStore.StoredMessage;
 import io.moquette.spi.ISessionsStore;
 import io.moquette.spi.MessageGUID;
@@ -27,6 +28,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.*;
+import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 
@@ -44,8 +46,6 @@ class MapDBSessionsStore implements ISessionsStore {
     //map clientID <-> set of currently in flight packet identifiers
     private Map<String, Set<Integer>> m_inFlightIds;
     private ConcurrentMap<String, PersistentSession> m_persistentSessions;
-    //maps clientID->[guid*], insertion order cares, it's queue
-    private ConcurrentMap<String, List<MessageGUID>> m_enqueuedStore;
     //maps clientID->[MessageId -> guid]
     private ConcurrentMap<String, Map<Integer, MessageGUID>> m_secondPhaseStore;
 
@@ -62,7 +62,6 @@ class MapDBSessionsStore implements ISessionsStore {
         m_inflightStore = m_db.getHashMap("inflight");
         m_inFlightIds = m_db.getHashMap("inflightPacketIDs");
         m_persistentSessions = m_db.getHashMap("sessions");
-        m_enqueuedStore = m_db.getHashMap("sessionQueue");
         m_secondPhaseStore = m_db.getHashMap("secondPhase");
     }
 
@@ -210,23 +209,13 @@ class MapDBSessionsStore implements ISessionsStore {
     }
 
     @Override
-    public void bindToDeliver(MessageGUID guid, String clientID) {
-        List<MessageGUID> guids = Utils.defaultGet(m_enqueuedStore, clientID, new ArrayList<MessageGUID>());
-        guids.add(guid);
-        m_enqueuedStore.put(clientID, guids);
-        m_messagesStore.incUsageCounter(guid);
+    public BlockingQueue<StoredMessage> queue(String clientID) {
+        return this.m_db.getQueue(clientID);
     }
 
     @Override
-    public Collection<MessageGUID> enqueued(String clientID) {
-        return Utils.defaultGet(m_enqueuedStore, clientID, new ArrayList<MessageGUID>());
-    }
-
-    @Override
-    public void removeEnqueued(String clientID, MessageGUID guid) {
-        List<MessageGUID> guids = Utils.defaultGet(m_enqueuedStore, clientID, new ArrayList<MessageGUID>());
-        guids.remove(guid);
-        m_enqueuedStore.put(clientID, guids);
+    public void dropQueue(String clientID) {
+        this.m_db.delete(clientID);
     }
 
     @Override
diff --git a/broker/src/test/java/io/moquette/spi/impl/MemoryMessagesStore.java b/broker/src/test/java/io/moquette/spi/impl/MemoryMessagesStore.java
index c24723b0..79814fed 100644
--- a/broker/src/test/java/io/moquette/spi/impl/MemoryMessagesStore.java
+++ b/broker/src/test/java/io/moquette/spi/impl/MemoryMessagesStore.java
@@ -78,15 +78,6 @@ public class MemoryMessagesStore implements IMessagesStore {
         return guid;
     }
 
-    @Override
-    public List<StoredMessage> listMessagesInSession(Collection<MessageGUID> guids) {
-        List<StoredMessage> ret = new ArrayList<>();
-        for (MessageGUID guid : guids) {
-            ret.add(m_persistentMessageStore.get(guid));
-        }
-        return ret;
-    }
-
     @Override
     public void dropMessagesInSession(String clientID) {
         Map<Integer, MessageGUID> messageGUIDMap = m_messageToGuids.get(clientID);
diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java
index df76ff2f..f2ee8aef 100644
--- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java
+++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java
@@ -18,7 +18,6 @@ package io.moquette.spi.impl;
 import io.moquette.parser.proto.messages.*;
 import io.moquette.interception.InterceptHandler;
 import io.moquette.parser.proto.messages.AbstractMessage.QOSType;
-import io.moquette.server.ConnectionDescriptor;
 import io.moquette.server.netty.NettyUtils;
 import io.moquette.spi.*;
 import io.moquette.spi.IMessagesStore.StoredMessage;
@@ -32,6 +31,8 @@ import org.junit.Test;
 
 import java.nio.ByteBuffer;
 import java.util.*;
+import java.util.concurrent.BlockingQueue;
+
 import static io.moquette.spi.impl.NettyChannelAssertions.assertConnAckAccepted;
 import static io.moquette.spi.impl.ProtocolProcessor.lowerQosToTheSubscriptionDesired;
 import static org.junit.Assert.*;
@@ -382,8 +383,8 @@ public class ProtocolProcessorTest {
         m_processor.processConnect(m_channel, connectMessage);
 
         //Verify no messages are still stored
-        Collection<MessageGUID> guids = m_sessionStore.enqueued(FAKE_PUBLISHER_ID);
-        assertTrue(m_messagesStore.listMessagesInSession(guids).isEmpty());
+        BlockingQueue<StoredMessage> messages = m_sessionStore.queue(FAKE_PUBLISHER_ID);
+        assertTrue(messages.isEmpty());
     }
     
     @Test
diff --git a/broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java b/broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java
index 892746bf..aefd83fb 100644
--- a/broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java
+++ b/broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java
@@ -15,6 +15,7 @@
  */
 package io.moquette.spi.persistence;
 
+import io.moquette.server.Constants;
 import io.moquette.spi.ClientSession;
 import io.moquette.spi.IMessagesStore;
 import io.moquette.spi.IMessagesStore.StoredMessage;
@@ -27,6 +28,8 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.*;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
 
 /**
  *
@@ -42,8 +45,8 @@ public class MemorySessionStore implements ISessionsStore {
     //maps clientID->[MessageId -> guid]
     private Map<String, Map<Integer, MessageGUID>> m_inflightStore = new HashMap<>();
 //    private Map<String, Set<Integer>> m_inflightIDs = new HashMap<>();
-    //maps clientID->[guid*]
-    private Map<String, Set<MessageGUID>> m_enqueuedStore = new HashMap<>();
+    //maps clientID->BlockingQueue
+    private Map<String, BlockingQueue<StoredMessage>> queues = new HashMap<>();
     //maps clientID->[MessageId -> guid]
     private Map<String, Map<Integer, MessageGUID>> m_secondPhaseStore = new HashMap<>();
 
@@ -205,22 +208,13 @@ public class MemorySessionStore implements ISessionsStore {
     }
 
     @Override
-    public void bindToDeliver(MessageGUID guid, String clientID) {
-        Set<MessageGUID> guids = Utils.defaultGet(m_enqueuedStore, clientID, new HashSet<MessageGUID>());
-        guids.add(guid);
-        m_enqueuedStore.put(clientID, guids);
+    public BlockingQueue<StoredMessage> queue(String clientID) {
+        return Utils.defaultGet(queues, clientID, new ArrayBlockingQueue<StoredMessage>(Constants.MAX_MESSAGE_QUEUE));
     }
 
     @Override
-    public Collection<MessageGUID> enqueued(String clientID) {
-        return Utils.defaultGet(m_enqueuedStore, clientID, new HashSet<MessageGUID>());
-    }
-
-    @Override
-    public void removeEnqueued(String clientID, MessageGUID guid) {
-        Set<MessageGUID> guids = Utils.defaultGet(m_enqueuedStore, clientID, new HashSet<MessageGUID>());
-        guids.remove(guid);
-        m_enqueuedStore.put(clientID, guids);
+    public void dropQueue(String clientID) {
+        queues.remove(clientID);
     }
 
     @Override