Moved avay from ConnectionDescriptorStore unused methods and stuff that hasn't nothing about ConnectionDescriptor storing

broker/src/main/java/io/moquette/connections/IConnectionsManager.java

@@ -16,10 +16,13 @@
 
 package io.moquette.connections;
 
+import io.moquette.server.ConnectionDescriptor;
+
 import java.util.Collection;
+import java.util.Optional;
 
 /**
- * This interface will be used by an external codebase to retrieve and close physical connections.
+ * Repository for active connections
  */
 public interface IConnectionsManager {
 
@@ -28,7 +31,15 @@ public interface IConnectionsManager {
      *
      * @return
      */
-    int getActiveConnectionsNo();
+    int countActiveConnections();
+
+    Optional<ConnectionDescriptor> lookupDescriptor(String clientID);
+
+    ConnectionDescriptor addConnection(ConnectionDescriptor descriptor);
+
+    boolean removeConnection(ConnectionDescriptor descriptor);
+
+    ConnectionDescriptor getConnection(String clientID);
 
     /**
      * Determines wether a MQTT client is connected to the broker.
@@ -44,21 +55,4 @@ public interface IConnectionsManager {
      * @return
      */
     Collection<String> getConnectedClientIds();
-
-    /**
-     * Closes a physical connection.
-     *
-     * @param clientID
-     * @param closeImmediately
-     *            If false, the connection will be flushed before it is closed.
-     * @return
-     */
-    boolean closeConnection(String clientID, boolean closeImmediately);
-
-    /**
-     * Returns the state of all the sessions
-     *
-     * @return
-     */
-    Collection<MqttSession> getSessions();
 }

broker/src/main/java/io/moquette/server/ConnectionDescriptor.java

@@ -69,19 +69,6 @@ public class ConnectionDescriptor {
         this.channel.writeAndFlush(payload).addListener(FIRE_EXCEPTION_ON_FAILURE);
     }
 
-    public void setupAutoFlusher(int flushIntervalMs) {
-        try {
-            this.channel.pipeline().addAfter(
-                    "idleEventHandler",
-                    "autoFlusher",
-                    new AutoFlushHandler(flushIntervalMs, TimeUnit.MILLISECONDS));
-        } catch (NoSuchElementException nseex) {
-            // the idleEventHandler is not present on the pipeline
-            this.channel.pipeline()
-                    .addFirst("autoFlusher", new AutoFlushHandler(flushIntervalMs, TimeUnit.MILLISECONDS));
-        }
-    }
-
     public boolean doesNotUseChannel(Channel channel) {
         return !(this.channel.equals(channel));
     }

broker/src/main/java/io/moquette/server/ConnectionDescriptorStore.java

@@ -17,21 +17,11 @@
 package io.moquette.server;
 
 import io.moquette.connections.IConnectionsManager;
-import io.moquette.connections.MqttConnectionMetrics;
-import io.moquette.connections.MqttSession;
-import io.moquette.connections.MqttSubscription;
-import io.moquette.server.netty.metrics.BytesMetrics;
-import io.moquette.server.netty.metrics.MessageMetrics;
-import io.moquette.spi.ClientSession;
-import io.moquette.spi.impl.SessionsRepository;
-import io.moquette.spi.impl.subscriptions.Subscription;
-import io.netty.handler.codec.mqtt.MqttMessage;
-import io.netty.handler.codec.mqtt.MqttMessageType;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 
@@ -40,56 +30,39 @@ public class ConnectionDescriptorStore implements IConnectionsManager {
     private static final Logger LOG = LoggerFactory.getLogger(ConnectionDescriptorStore.class);
 
     private final ConcurrentMap<String, ConnectionDescriptor> connectionDescriptors;
-    private final SessionsRepository sessionsRepository;
 
-    public ConnectionDescriptorStore(SessionsRepository sessionsRepository) {
+    public ConnectionDescriptorStore() {
         this.connectionDescriptors = new ConcurrentHashMap<>();
-        this.sessionsRepository = sessionsRepository;
     }
 
-    public boolean sendMessage(MqttMessage message, Integer messageID, String clientID) {
+    @Override
-        final MqttMessageType messageType = message.fixedHeader().messageType();
+    public Optional<ConnectionDescriptor> lookupDescriptor(String clientID) {
-        try {
+        if (clientID == null) {
-            if (messageID != null) {
+            return Optional.empty();
-                LOG.info("Sending {} message CId=<{}>, messageId={}", messageType, clientID, messageID);
-            } else {
-                LOG.debug("Sending {} message CId=<{}>", messageType, clientID);
-            }
-
-            ConnectionDescriptor descriptor = connectionDescriptors.get(clientID);
-            if (descriptor == null) {
-                if (messageID != null) {
-                    LOG.error("Client has just disconnected. {} message could not be sent. CId=<{}>, messageId={}",
-                        messageType, clientID, messageID);
-                } else {
-                    LOG.error("Client has just disconnected. {} could not be sent. CId=<{}>", messageType, clientID);
-                }
-                /*
-                 * If the client has just disconnected, its connection descriptor will be null. We
-                 * don't have to make the broker crash: we'll just discard the PUBACK message.
-                 */
-                return false;
-            }
-            descriptor.writeAndFlush(message);
-            return true;
-        } catch (Throwable e) {
-            String errorMsg = "Unable to send " + messageType + " message. CId=<" + clientID + ">";
-            if (messageID != null) {
-                errorMsg += ", messageId=" + messageID;
-            }
-            LOG.error(errorMsg, e);
-            return false;
         }
+
+        ConnectionDescriptor descriptor = connectionDescriptors.get(clientID);
+        if (descriptor == null) {
+            /*
+             * If the client has just disconnected, its connection descriptor will be null. We
+             * don't have to make the broker crash: we'll just discard the PUBACK message.
+             */
+            return Optional.empty();
+        }
+        return Optional.of(descriptor);
     }
 
+    @Override
     public ConnectionDescriptor addConnection(ConnectionDescriptor descriptor) {
         return connectionDescriptors.putIfAbsent(descriptor.clientID, descriptor);
     }
 
+    @Override
     public boolean removeConnection(ConnectionDescriptor descriptor) {
         return connectionDescriptors.remove(descriptor.clientID, descriptor);
     }
 
+    @Override
     public ConnectionDescriptor getConnection(String clientID) {
         return connectionDescriptors.get(clientID);
     }
@@ -100,7 +73,7 @@ public class ConnectionDescriptorStore implements IConnectionsManager {
     }
 
     @Override
-    public int getActiveConnectionsNo() {
+    public int countActiveConnections() {
         return connectionDescriptors.size();
     }
 
@@ -108,56 +81,4 @@ public class ConnectionDescriptorStore implements IConnectionsManager {
     public Collection<String> getConnectedClientIds() {
         return connectionDescriptors.keySet();
     }
-
-    @Override
-    public boolean closeConnection(String clientID, boolean closeImmediately) {
-        ConnectionDescriptor descriptor = connectionDescriptors.get(clientID);
-        if (descriptor == null) {
-            LOG.error("Connection descriptor doesn't exist. MQTT connection cannot be closed. CId=<{}>, " +
-                "closeImmediately={}", clientID, closeImmediately);
-            return false;
-        }
-        if (closeImmediately) {
-            descriptor.abort();
-            return true;
-        } else {
-            return descriptor.close();
-        }
-    }
-
-    @Override
-    public Collection<MqttSession> getSessions() {
-        LOG.info("Retrieving status of all sessions.");
-        Collection<MqttSession> result = new ArrayList<>();
-        for (ClientSession session : this.sessionsRepository.getAllSessions()) {
-            result.add(buildMqttSession(session));
-        }
-        return result;
-    }
-
-    private MqttSession buildMqttSession(ClientSession session) {
-        MqttSession result = new MqttSession();
-        Collection<MqttSubscription> mqttSubscriptions = new ArrayList<>();
-        for (Subscription subscription : session.getSubscriptions()) {
-            mqttSubscriptions.add(new MqttSubscription(subscription.getRequestedQos().toString(),
-                subscription.getClientId(), subscription.getTopicFilter().toString(), subscription.isActive()));
-        }
-        result.setActiveSubscriptions(mqttSubscriptions);
-        result.setCleanSession(session.isCleanSession());
-        ConnectionDescriptor descriptor = this.getConnection(session.clientID);
-        if (descriptor != null) {
-            result.setConnectionEstablished(true);
-            BytesMetrics bytesMetrics = descriptor.getBytesMetrics();
-            MessageMetrics messageMetrics = descriptor.getMessageMetrics();
-            result.setConnectionMetrics(new MqttConnectionMetrics(bytesMetrics.readBytes(), bytesMetrics.wroteBytes(),
-                messageMetrics.messagesRead(), messageMetrics.messagesWrote()));
-        } else {
-            result.setConnectionEstablished(false);
-        }
-        result.setPendingPublishMessagesNo(session.getPendingPublishMessagesNo());
-        result.setSecondPhaseAckPendingMessages(session.countPubReleaseWaitingPubComplete());
-        result.setInflightMessages(session.getInflightMessagesNo());
-        return result;
-    }
-
 }

broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java

@@ -16,7 +16,7 @@
 
 package io.moquette.spi.impl;
 
-import io.moquette.server.ConnectionDescriptorStore;
+import io.moquette.connections.IConnectionsManager;
 import io.moquette.spi.ClientSession;
 import io.moquette.spi.IMessagesStore;
 import io.moquette.spi.impl.subscriptions.ISubscriptionsDirectory;
@@ -34,21 +34,21 @@ import static io.moquette.spi.impl.ProtocolProcessor.lowerQosToTheSubscriptionDe
 class MessagesPublisher {
 
     private static final Logger LOG = LoggerFactory.getLogger(MessagesPublisher.class);
-    private final ConnectionDescriptorStore connectionDescriptors;
+    private final IConnectionsManager connectionDescriptors;
     private final PersistentQueueMessageSender messageSender;
     private final ISubscriptionsDirectory subscriptions;
     private SessionsRepository sessionsRepository;
 
-    public MessagesPublisher(ConnectionDescriptorStore connectionDescriptors,
+    MessagesPublisher(IConnectionsManager connectionDescriptors,
-                             PersistentQueueMessageSender messageSender, ISubscriptionsDirectory subscriptions,
+                      PersistentQueueMessageSender messageSender, ISubscriptionsDirectory subscriptions,
-                             SessionsRepository sessionsRepository) {
+                      SessionsRepository sessionsRepository) {
         this.connectionDescriptors = connectionDescriptors;
         this.messageSender = messageSender;
         this.subscriptions = subscriptions;
         this.sessionsRepository = sessionsRepository;
     }
 
-    static MqttPublishMessage notRetainedPublish(String topic, MqttQoS qos, ByteBuf message) {
+    private static MqttPublishMessage notRetainedPublish(String topic, MqttQoS qos, ByteBuf message) {
         return notRetainedPublishWithMessageId(topic, qos, message, 0);
     }
 

broker/src/main/java/io/moquette/spi/impl/PersistentQueueMessageSender.java

@@ -16,21 +16,27 @@
 
 package io.moquette.spi.impl;
 
+import io.moquette.connections.IConnectionsManager;
+import io.moquette.server.ConnectionDescriptor;
 import io.moquette.server.ConnectionDescriptorStore;
 import io.moquette.spi.ClientSession;
+import io.netty.handler.codec.mqtt.MqttMessageType;
 import io.netty.handler.codec.mqtt.MqttPublishMessage;
 import io.netty.handler.codec.mqtt.MqttQoS;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+
+import java.util.Optional;
+
 import static io.moquette.spi.impl.ProtocolProcessor.asStoredMessage;
 import static io.netty.handler.codec.mqtt.MqttQoS.AT_MOST_ONCE;
 
 class PersistentQueueMessageSender {
 
     private static final Logger LOG = LoggerFactory.getLogger(PersistentQueueMessageSender.class);
-    private final ConnectionDescriptorStore connectionDescriptorStore;
+    private final IConnectionsManager connectionDescriptorStore;
 
-    PersistentQueueMessageSender(ConnectionDescriptorStore connectionDescriptorStore) {
+    PersistentQueueMessageSender(IConnectionsManager connectionDescriptorStore) {
         this.connectionDescriptorStore = connectionDescriptorStore;
     }
 
@@ -46,7 +52,19 @@ class PersistentQueueMessageSender {
             LOG.info("Sending PUBLISH message. MessageId={}, CId={}, topic={}", messageId, clientId, topicName);
         }
 
-        boolean messageDelivered = connectionDescriptorStore.sendMessage(pubMessage, messageId, clientId);
+        boolean messageDelivered = false;
+
+        final Optional<ConnectionDescriptor> optDescriptor = connectionDescriptorStore.lookupDescriptor(clientId);
+        if (optDescriptor.isPresent()) {
+            final ConnectionDescriptor descriptor = optDescriptor.get();
+            try {
+                descriptor.writeAndFlush(pubMessage);
+                messageDelivered = true;
+            } catch (Throwable e) {
+                LOG.error("Unable to send {} message. CId=<{}>, messageId={}", pubMessage.fixedHeader().messageType(),
+                          clientId, messageId, e);
+            }
+        }
 
         if (!messageDelivered) {
             if (qos != AT_MOST_ONCE && !clientsession.isCleanSession()) {

broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java

@@ -16,10 +16,12 @@
 
 package io.moquette.spi.impl;
 
+import io.moquette.connections.IConnectionsManager;
 import io.moquette.interception.InterceptHandler;
 import io.moquette.interception.messages.InterceptAcknowledgedMessage;
 import io.moquette.server.ConnectionDescriptor;
 import io.moquette.server.ConnectionDescriptorStore;
+import io.moquette.server.netty.AutoFlushHandler;
 import io.moquette.server.netty.NettyUtils;
 import io.moquette.spi.*;
 import io.moquette.spi.IMessagesStore.StoredMessage;
@@ -41,6 +43,7 @@ import java.nio.charset.StandardCharsets;
 import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.TimeUnit;
 
 import static io.moquette.server.ConnectionDescriptor.ConnectionState.*;
 import static io.moquette.spi.impl.InternalRepublisher.createPublishForQos;
@@ -97,11 +100,11 @@ public class ProtocolProcessor {
 
     private static final Logger LOG = LoggerFactory.getLogger(ProtocolProcessor.class);
 
-    private ConnectionDescriptorStore connectionDescriptors;
+    private IConnectionsManager connectionDescriptors;
     private ConcurrentMap<RunningSubscription, SubscriptionState> subscriptionInCourse;
 
     private ISubscriptionsDirectory subscriptions;
-    private ISubscriptionsStore subscriptionStore;
+//    private ISubscriptionsStore subscriptionStore;
     private boolean allowAnonymous;
     private boolean allowZeroByteClientId;
     private IAuthorizator m_authorizator;
@@ -136,7 +139,7 @@ public class ProtocolProcessor {
     public void init(ISubscriptionsDirectory subscriptions, IMessagesStore storageService, ISessionsStore sessionsStore,
                      IAuthenticator authenticator, boolean allowAnonymous, boolean allowZeroByteClientId,
                      IAuthorizator authorizator, BrokerInterceptor interceptor, SessionsRepository sessionsRepository) {
-        init(new ConnectionDescriptorStore(sessionsRepository), subscriptions, storageService, sessionsStore,
+        init(new ConnectionDescriptorStore(), subscriptions, storageService, sessionsStore,
              authenticator, allowAnonymous, allowZeroByteClientId, authorizator, interceptor, sessionsRepository);
     }
 
@@ -158,11 +161,11 @@ public class ProtocolProcessor {
      * @param interceptor
      *            to notify events to an intercept handler
      */
-    void init(ConnectionDescriptorStore connectionDescriptors, ISubscriptionsDirectory subscriptions,
+    void init(IConnectionsManager connectionDescriptors, ISubscriptionsDirectory subscriptions,
               IMessagesStore storageService, ISessionsStore sessionsStore, IAuthenticator authenticator,
               boolean allowAnonymous, boolean allowZeroByteClientId, IAuthorizator authorizator,
               BrokerInterceptor interceptor, SessionsRepository sessionsRepository) {
-        LOG.info("Initializing MQTT protocol processor...");
+        LOG.debug("Initializing MQTT protocol processor...");
         this.connectionDescriptors = connectionDescriptors;
         this.subscriptionInCourse = new ConcurrentHashMap<>();
         this.m_interceptor = interceptor;
@@ -176,7 +179,6 @@ public class ProtocolProcessor {
         m_authenticator = authenticator;
         m_messagesStore = storageService;
         m_sessionsStore = sessionsStore;
-        subscriptionStore = sessionsStore.subscriptionStore();
 
         this.sessionsRepository = sessionsRepository;
 
@@ -185,7 +187,7 @@ public class ProtocolProcessor {
         this.messagesPublisher = new MessagesPublisher(connectionDescriptors, messageSender,
             subscriptions, this.sessionsRepository);
 
-        LOG.info("Initializing QoS publish handlers...");
+        LOG.debug("Initializing QoS publish handlers...");
         this.qos0PublishHandler = new Qos0PublishHandler(m_authorizator, m_messagesStore, m_interceptor,
                 this.messagesPublisher);
         this.qos1PublishHandler = new Qos1PublishHandler(m_authorizator, m_messagesStore, m_interceptor,
@@ -193,7 +195,7 @@ public class ProtocolProcessor {
         this.qos2PublishHandler = new Qos2PublishHandler(m_authorizator, subscriptions, m_messagesStore, m_interceptor,
                 this.connectionDescriptors, this.messagesPublisher, this.sessionsRepository);
 
-        LOG.info("Initializing internal republisher...");
+        LOG.debug("Initializing internal republisher...");
         this.internalRepublisher = new InternalRepublisher(messageSender);
     }
 
@@ -263,6 +265,10 @@ public class ProtocolProcessor {
             channel.close().addListener(CLOSE_ON_FAILURE);
             return;
         }
+
+        int flushIntervalMs = 500/* (keepAlive * 1000) / 2 */;
+        setupAutoFlusher(channel, flushIntervalMs);
+
         final boolean success = descriptor.assignState(MESSAGES_REPUBLISHED, ESTABLISHED);
         if (!success) {
             channel.close().addListener(CLOSE_ON_FAILURE);
@@ -271,6 +277,19 @@ public class ProtocolProcessor {
         LOG.info("Connected client <{}> with login <{}>", clientId, payload.userName());
     }
 
+    private void setupAutoFlusher(Channel channel, int flushIntervalMs) {
+        try {
+            channel.pipeline().addAfter(
+                "idleEventHandler",
+                "autoFlusher",
+                new AutoFlushHandler(flushIntervalMs, TimeUnit.MILLISECONDS));
+        } catch (NoSuchElementException nseex) {
+            // the idleEventHandler is not present on the pipeline
+            channel.pipeline()
+                .addFirst("autoFlusher", new AutoFlushHandler(flushIntervalMs, TimeUnit.MILLISECONDS));
+        }
+    }
+
     private MqttConnAckMessage connAck(MqttConnectReturnCode returnCode) {
         return connAck(returnCode, false);
     }
@@ -379,8 +398,6 @@ public class ProtocolProcessor {
             LOG.info("Republishing stored publish events. CId={}", clientSession.clientID);
             this.internalRepublisher.publishStored(clientSession);
         }
-        int flushIntervalMs = 500/* (keepAlive * 1000) / 2 */;
-        descriptor.setupAutoFlusher(flushIntervalMs);
         return true;
     }
 

broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java

@@ -165,7 +165,7 @@ public class ProtocolProcessorBootstrapper {
         }
 
         LOG.info("Initializing connection descriptor store...");
-        connectionDescriptors = new ConnectionDescriptorStore(sessionsRepository);
+        connectionDescriptors = new ConnectionDescriptorStore();
 
         LOG.info("Initializing MQTT protocol processor...");
         boolean allowAnonymous = Boolean

broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java

@@ -16,6 +16,8 @@
 
 package io.moquette.spi.impl;
 
+import io.moquette.connections.IConnectionsManager;
+import io.moquette.server.ConnectionDescriptor;
 import io.moquette.server.ConnectionDescriptorStore;
 import io.moquette.server.netty.NettyUtils;
 import io.moquette.spi.IMessagesStore;
@@ -29,6 +31,8 @@ import io.netty.handler.codec.mqtt.MqttPublishMessage;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.Optional;
+
 import static io.moquette.spi.impl.ProtocolProcessor.asStoredMessage;
 import static io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader.from;
 import static io.netty.handler.codec.mqtt.MqttQoS.AT_MOST_ONCE;
@@ -39,11 +43,11 @@ class Qos1PublishHandler extends QosPublishHandler {
 
     private final IMessagesStore m_messagesStore;
     private final BrokerInterceptor m_interceptor;
-    private final ConnectionDescriptorStore connectionDescriptors;
+    private final IConnectionsManager connectionDescriptors;
     private final MessagesPublisher publisher;
 
-    public Qos1PublishHandler(IAuthorizator authorizator, IMessagesStore messagesStore, BrokerInterceptor interceptor,
+    Qos1PublishHandler(IAuthorizator authorizator, IMessagesStore messagesStore, BrokerInterceptor interceptor,
-                              ConnectionDescriptorStore connectionDescriptors, MessagesPublisher messagesPublisher) {
+                       IConnectionsManager connectionDescriptors, MessagesPublisher messagesPublisher) {
         super(authorizator);
         this.m_messagesStore = messagesStore;
         this.m_interceptor = interceptor;
@@ -89,16 +93,17 @@ class Qos1PublishHandler extends QosPublishHandler {
         MqttPubAckMessage pubAckMessage = new MqttPubAckMessage(fixedHeader, from(messageID));
 
         try {
-            if (connectionDescriptors == null) {
-                throw new RuntimeException("Internal bad error, found connectionDescriptors to null while it should " +
-                    "be initialized, somewhere it's overwritten!!");
-            }
             LOG.trace("connected clientIDs are {}", connectionDescriptors.getConnectedClientIds());
             if (!connectionDescriptors.isConnected(clientId)) {
                 throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client %s in cache %s",
                     clientId, connectionDescriptors));
             }
-            connectionDescriptors.sendMessage(pubAckMessage, messageID, clientId);
+            LOG.info("Sending {} message CId=<{}>", MqttMessageType.PUBACK, clientId);
+            final Optional<ConnectionDescriptor> optDescriptor = connectionDescriptors.lookupDescriptor(clientId);
+            if (optDescriptor.isPresent()) {
+                final ConnectionDescriptor descriptor = optDescriptor.get();
+                descriptor.writeAndFlush(pubAckMessage);
+            }
         } catch (Throwable t) {
             LOG.error(null, t);
         }

broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java

@@ -16,6 +16,8 @@
 
 package io.moquette.spi.impl;
 
+import io.moquette.connections.IConnectionsManager;
+import io.moquette.server.ConnectionDescriptor;
 import io.moquette.server.ConnectionDescriptorStore;
 import io.moquette.server.netty.NettyUtils;
 import io.moquette.spi.ClientSession;
@@ -29,6 +31,8 @@ import io.netty.handler.codec.mqtt.*;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.Optional;
+
 import static io.moquette.spi.impl.DebugUtils.payload2Str;
 import static io.moquette.spi.impl.ProtocolProcessor.asStoredMessage;
 import static io.moquette.spi.impl.Utils.messageId;
@@ -42,14 +46,14 @@ class Qos2PublishHandler extends QosPublishHandler {
     private final ISubscriptionsDirectory subscriptions;
     private final IMessagesStore m_messagesStore;
     private final BrokerInterceptor m_interceptor;
-    private final ConnectionDescriptorStore connectionDescriptors;
+    private final IConnectionsManager connectionDescriptors;
     private final MessagesPublisher publisher;
     private final SessionsRepository sessionsRepository;
 
-    public Qos2PublishHandler(IAuthorizator authorizator, ISubscriptionsDirectory subscriptions,
+    Qos2PublishHandler(IAuthorizator authorizator, ISubscriptionsDirectory subscriptions,
-                              IMessagesStore messagesStore, BrokerInterceptor interceptor,
+                       IMessagesStore messagesStore, BrokerInterceptor interceptor,
-                              ConnectionDescriptorStore connectionDescriptors,
+                       IConnectionsManager connectionDescriptors,
-                              MessagesPublisher messagesPublisher, SessionsRepository sessionsRepository) {
+                       MessagesPublisher messagesPublisher, SessionsRepository sessionsRepository) {
         super(authorizator);
         this.subscriptions = subscriptions;
         this.m_messagesStore = messagesStore;
@@ -129,16 +133,34 @@ class Qos2PublishHandler extends QosPublishHandler {
     }
 
     private void sendPubRec(String clientID, int messageID) {
-        LOG.debug("Sending PUBREC message. CId={}, messageId={}", clientID, messageID);
+        LOG.trace("Sending PUBREC message. CId={}, messageId={}", clientID, messageID);
         MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBREC, false, AT_MOST_ONCE, false, 0);
         MqttMessage pubRecMessage = new MqttMessage(fixedHeader, from(messageID));
-        connectionDescriptors.sendMessage(pubRecMessage, messageID, clientID);
+        final Optional<ConnectionDescriptor> connectionDescriptor = connectionDescriptors.lookupDescriptor(clientID);
+        if (connectionDescriptor.isPresent()) {
+            final ConnectionDescriptor descriptor = connectionDescriptor.get();
+            try {
+                descriptor.writeAndFlush(pubRecMessage);
+            } catch (Throwable e) {
+                LOG.error("Unable to send {} message. CId=<{}>, messageId={}", MqttMessageType.PUBREC, clientID,
+                          messageID, e);
+            }
+        }
     }
 
     private void sendPubComp(String clientID, int messageID) {
-        LOG.debug("Sending PUBCOMP message. CId={}, messageId={}", clientID, messageID);
+        LOG.trace("Sending PUBCOMP message. CId={}, messageId={}", clientID, messageID);
         MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBCOMP, false, AT_MOST_ONCE, false, 0);
         MqttMessage pubCompMessage = new MqttMessage(fixedHeader, from(messageID));
-        connectionDescriptors.sendMessage(pubCompMessage, messageID, clientID);
+        final Optional<ConnectionDescriptor> connectionDescriptor = connectionDescriptors.lookupDescriptor(clientID);
+        if (connectionDescriptor.isPresent()) {
+            final ConnectionDescriptor descriptor = connectionDescriptor.get();
+            try {
+                descriptor.writeAndFlush(pubCompMessage);
+            } catch (Throwable e) {
+                LOG.error("Unable to send {} message. CId=<{}>, messageId={}", MqttMessageType.PUBCOMP, clientID,
+                          messageID, e);
+            }
+        }
     }
 }