diff --git a/broker/src/main/java/io/moquette/connections/IConnectionsManager.java b/broker/src/main/java/io/moquette/connections/IConnectionsManager.java index da20d252..18953fde 100644 --- a/broker/src/main/java/io/moquette/connections/IConnectionsManager.java +++ b/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 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 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 getSessions(); } diff --git a/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java b/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java index 755cddb0..fddced0b 100644 --- a/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java +++ b/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)); } diff --git a/broker/src/main/java/io/moquette/server/ConnectionDescriptorStore.java b/broker/src/main/java/io/moquette/server/ConnectionDescriptorStore.java index 209ee96c..22f8c832 100644 --- a/broker/src/main/java/io/moquette/server/ConnectionDescriptorStore.java +++ b/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 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) { - final MqttMessageType messageType = message.fixedHeader().messageType(); - try { - if (messageID != null) { - 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; + @Override + public Optional lookupDescriptor(String clientID) { + if (clientID == null) { + return Optional.empty(); } + + 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 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 getSessions() { - LOG.info("Retrieving status of all sessions."); - Collection 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 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; - } - } 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 efaf2ec1..d14a8b97 100644 --- a/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java +++ b/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, - PersistentQueueMessageSender messageSender, ISubscriptionsDirectory subscriptions, - SessionsRepository sessionsRepository) { + MessagesPublisher(IConnectionsManager connectionDescriptors, + PersistentQueueMessageSender messageSender, ISubscriptionsDirectory subscriptions, + 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); } 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 f469d52f..1485ec6d 100644 --- a/broker/src/main/java/io/moquette/spi/impl/PersistentQueueMessageSender.java +++ b/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 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()) { 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 63aac13e..76aaef2d 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/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 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; } diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java index 13d20871..da861f79 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java +++ b/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 diff --git a/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java b/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java index f7c5c9dc..69e921d0 100644 --- a/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java +++ b/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, - ConnectionDescriptorStore connectionDescriptors, MessagesPublisher messagesPublisher) { + Qos1PublishHandler(IAuthorizator authorizator, IMessagesStore messagesStore, BrokerInterceptor interceptor, + 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 optDescriptor = connectionDescriptors.lookupDescriptor(clientId); + if (optDescriptor.isPresent()) { + final ConnectionDescriptor descriptor = optDescriptor.get(); + descriptor.writeAndFlush(pubAckMessage); + } } catch (Throwable t) { LOG.error(null, t); } diff --git a/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java b/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java index 424b2410..71d764b1 100644 --- a/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java +++ b/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, - IMessagesStore messagesStore, BrokerInterceptor interceptor, - ConnectionDescriptorStore connectionDescriptors, - MessagesPublisher messagesPublisher, SessionsRepository sessionsRepository) { + Qos2PublishHandler(IAuthorizator authorizator, ISubscriptionsDirectory subscriptions, + IMessagesStore messagesStore, BrokerInterceptor interceptor, + IConnectionsManager connectionDescriptors, + 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 = 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 = 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); + } + } } }