diff --git a/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java b/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java index 5846d8f8..6e3c5c62 100644 --- a/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java +++ b/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java @@ -16,10 +16,16 @@ package io.moquette.server; import io.netty.channel.Channel; +import java.util.NoSuchElementException; +import java.util.concurrent.TimeUnit; + import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.concurrent.atomic.AtomicReference; +import io.moquette.server.netty.AutoFlushHandler; +import io.moquette.server.netty.NettyUtils; + /** * Value object to maintain the information of single connection, like ClientID, Channel, @@ -40,7 +46,7 @@ public class ConnectionDescriptor { } public final String clientID; - public final Channel channel; + private final Channel channel; public final boolean cleanSession; private final AtomicReference channelState = new AtomicReference<>(ConnectionState.DISCONNECTED); @@ -49,20 +55,59 @@ public class ConnectionDescriptor { this.channel = session; this.cleanSession = cleanSession; } + + public void writeAndFlush(Object payload) { + this.channel.writeAndFlush(payload); + } + + 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)); + } + + public boolean close() { + LOG.info("Closing connection descriptor. MqttClientId = {}.", clientID); + final boolean success = assignState(ConnectionState.INTERCEPTORS_NOTIFIED, ConnectionState.DISCONNECTED); + if (!success) { + return false; + } + this.channel.close(); + return true; + } + + public String getUsername() { + return NettyUtils.userName(this.channel); + } public void abort() { - LOG.info("closing the channel"); + LOG.info("Closing connection descriptor. MqttClientId = {}.", clientID); // try { //this.channel.disconnect().sync(); - this.channel.close();//.sync(); + this.channel.close();//.sync(); // } catch (InterruptedException e) { // e.printStackTrace(); // } } public boolean assignState(ConnectionState expected, ConnectionState newState) { - return channelState.compareAndSet(expected, newState); - } + LOG.debug("Updating state of connection descriptor. MqttClientId = {}, expectedState = {}, newState = {}.", + clientID, expected, newState); + boolean retval = channelState.compareAndSet(expected, newState); + if (!retval) { + LOG.error( + "Unable to update state of connection descriptor. MqttclientId = {}, expectedState = {}, newState = {}.", + clientID, expected, newState); + } + return retval; + } @Override public String toString() { diff --git a/broker/src/main/java/io/moquette/server/ConnectionDescriptorStore.java b/broker/src/main/java/io/moquette/server/ConnectionDescriptorStore.java new file mode 100644 index 00000000..4030c292 --- /dev/null +++ b/broker/src/main/java/io/moquette/server/ConnectionDescriptorStore.java @@ -0,0 +1,85 @@ +package io.moquette.server; + +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import io.moquette.parser.proto.messages.AbstractMessage; + +public class ConnectionDescriptorStore { + + private static final Logger LOG = LoggerFactory.getLogger(ConnectionDescriptorStore.class); + + private final ConcurrentMap connectionDescriptors; + + public ConnectionDescriptorStore() { + this.connectionDescriptors = new ConcurrentHashMap(); + } + + public boolean sendMessage(AbstractMessage message, String clientId) { + return sendMessage(message, null, clientId); + } + + public boolean sendMessage(AbstractMessage message, Integer messageID, String clientID) { + try { + // The connection descriptors map will never be null: it's a final + // attribute. + + if (messageID != null) { + LOG.info("Sending {} message. MqttClientId = {}, messageId = {}.", message.getMessageType(), clientID, + messageID); + } else { + LOG.debug("Sending {} message. MqttClientId = {}.", message.getMessageType(), clientID); + } + + ConnectionDescriptor descriptor = connectionDescriptors.get(clientID); + if (descriptor == null) { + if (messageID != null) { + LOG.error( + "The client has just disconnected. The {} message could not be sent. MqttClientId = {}, messageId = {}.", + message.getMessageType(), clientID, messageID); + } else { + LOG.error("The client has just disconnected. The {} could not be sent. MqttClientId = {}.", + message.getMessageType(), clientID); + } + /* + * If the client hast 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) { + if (messageID != null) { + LOG.error( + "Unable to send {} message. MqttClientId = {}, messageId = {}, cause = {}, errorMessage = {}.", + message.getMessageType(), clientID, messageID, e.getCause(), e.getMessage()); + } else { + LOG.error("Unable to send {} message. MqttClientId = {}, cause = {}, errorMessage = {}.", + message.getMessageType(), clientID, e.getCause(), e.getMessage()); + } + return false; + } + } + + public ConnectionDescriptor addConnection(ConnectionDescriptor descriptor) { + return connectionDescriptors.putIfAbsent(descriptor.clientID, descriptor); + } + + public boolean removeConnection(ConnectionDescriptor descriptor) { + return connectionDescriptors.remove(descriptor.clientID, descriptor); + } + + public ConnectionDescriptor getConnection(String clientID) { + return connectionDescriptors.get(clientID); + } + + public boolean isConnected(String clientID) { + return connectionDescriptors.containsKey(clientID); + } + +} 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 5f000d0d..95988f6c 100644 --- a/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java +++ b/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java @@ -2,7 +2,7 @@ package io.moquette.spi.impl; import io.moquette.parser.proto.messages.AbstractMessage; import io.moquette.parser.proto.messages.PublishMessage; -import io.moquette.server.ConnectionDescriptor; +import io.moquette.server.ConnectionDescriptorStore; import io.moquette.spi.ClientSession; import io.moquette.spi.IMessagesStore; import io.moquette.spi.ISessionsStore; @@ -13,19 +13,18 @@ import org.slf4j.LoggerFactory; import java.nio.ByteBuffer; import java.util.List; -import java.util.concurrent.ConcurrentMap; import static io.moquette.spi.impl.ProtocolProcessor.lowerQosToTheSubscriptionDesired; class MessagesPublisher { private static final Logger LOG = LoggerFactory.getLogger(MessagesPublisher.class); - private final ConcurrentMap connectionDescriptors; + private final ConnectionDescriptorStore connectionDescriptors; private final ISessionsStore m_sessionsStore; private final IMessagesStore m_messagesStore; private final PersistentQueueMessageSender messageSender; - public MessagesPublisher(ConcurrentMap connectionDescriptors, ISessionsStore sessionsStore, + public MessagesPublisher(ConnectionDescriptorStore connectionDescriptors, ISessionsStore sessionsStore, IMessagesStore messagesStore, PersistentQueueMessageSender messageSender) { this.connectionDescriptors = connectionDescriptors; this.m_sessionsStore = sessionsStore; @@ -59,7 +58,7 @@ class MessagesPublisher { AbstractMessage.QOSType qos = lowerQosToTheSubscriptionDesired(sub, publishingQos); ClientSession targetSession = m_sessionsStore.sessionForClient(sub.getClientId()); - boolean targetIsActive = this.connectionDescriptors.containsKey(sub.getClientId()); + boolean targetIsActive = this.connectionDescriptors.isConnected(sub.getClientId()); LOG.debug("Broker republishing to client <{}> topicFilter <{}> qos <{}>, active {}", sub.getClientId(), sub.getTopicFilter(), qos, targetIsActive); 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 30f5bae2..ebbf7e69 100644 --- a/broker/src/main/java/io/moquette/spi/impl/PersistentQueueMessageSender.java +++ b/broker/src/main/java/io/moquette/spi/impl/PersistentQueueMessageSender.java @@ -1,13 +1,11 @@ package io.moquette.spi.impl; import io.moquette.parser.proto.messages.PublishMessage; -import io.moquette.server.ConnectionDescriptor; +import io.moquette.server.ConnectionDescriptorStore; import io.moquette.spi.ClientSession; -import io.netty.channel.Channel; import org.slf4j.Logger; 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; @@ -15,41 +13,34 @@ import static io.moquette.spi.impl.ProtocolProcessor.asStoredMessage; class PersistentQueueMessageSender { private static final Logger LOG = LoggerFactory.getLogger(PersistentQueueMessageSender.class); - private final ConcurrentMap connectionDescriptors; + private final ConnectionDescriptorStore connectionDescriptorStore; - public PersistentQueueMessageSender(ConcurrentMap connectionDescriptors) { - this.connectionDescriptors = connectionDescriptors; + public PersistentQueueMessageSender(ConnectionDescriptorStore connectionDescriptorStore) { + this.connectionDescriptorStore = connectionDescriptorStore; } void sendPublish(ClientSession clientsession, PublishMessage pubMessage) { String clientId = clientsession.clientID; - LOG.info("send publish message to <{}> on topic <{}>", clientId, pubMessage.getTopicName()); if (LOG.isDebugEnabled()) { - LOG.debug("directSend invoked clientId <{}> on topic <{}> QoS {} retained {} messageID {}", - clientId, pubMessage.getTopicName(), pubMessage.getQos(), false, pubMessage.getMessageID()); - LOG.debug("content <{}>", DebugUtils.payload2Str(pubMessage.getPayload())); + LOG.debug("Sending PUBLISH message. MessageId = {}, mqttClientId = {}, topic = {}, qos = {}, payload = {}.", + pubMessage.getMessageID(), + clientId, pubMessage.getTopicName(), DebugUtils.payload2Str(pubMessage.getPayload())); + } else { + LOG.info("Sending PUBLISH message. MessageId = {}, mqttClientId = {}, topic = {}.", pubMessage.getMessageID(), + clientId, pubMessage.getTopicName()); } - if (connectionDescriptors == null) { - throw new RuntimeException("Internal bad error, found connectionDescriptors to null while it should be " + - "initialized, somewhere it's overwritten!!"); - } - if (connectionDescriptors.get(clientId) == null) { - //TODO while we were publishing to the target client, that client disconnected, - // could happen is not an error HANDLE IT - throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client <%s> in cache <%s>", - clientId, connectionDescriptors)); - } - Channel channel = connectionDescriptors.get(clientId).channel; - LOG.trace("Session for clientId {}", clientId); - if (channel.isWritable()) { - LOG.debug("channel is writable"); - //if channel is writable don't enqueue - channel.writeAndFlush(pubMessage); - } else if (pubMessage.getQos() != MOST_ONE) { - //enqueue to the client session - LOG.debug("enqueue to client session"); - clientsession.enqueue(asStoredMessage(pubMessage)); - } + boolean messageDelivered = connectionDescriptorStore.sendMessage(pubMessage, pubMessage.getMessageID(), clientId); + + if (!messageDelivered && pubMessage.getQos() != MOST_ONE && !clientsession.isCleanSession()) { + LOG.warn( + "The PUBLISH message could not be delivered. It will be stored. MessageId = {}, mqttClientId = {}, topic = {}, qos = {}, cleanSession = {}.", + pubMessage.getMessageID(), clientId, pubMessage.getTopicName(), pubMessage.getQos(), false); + clientsession.enqueue(asStoredMessage(pubMessage)); + } else { + LOG.warn( + "The PUBLISH message could not be delivered. It will be discarded. MessageId = {}, mqttClientId = {}, topic = {}, qos = {}, cleanSession = {}.", + pubMessage.getMessageID(), clientId, pubMessage.getTopicName(), pubMessage.getQos(), true); + } } } 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 008953c8..a8913ee4 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -23,7 +23,7 @@ import io.moquette.interception.InterceptHandler; import io.moquette.parser.proto.messages.*; import io.moquette.server.ConnectionDescriptor; import io.moquette.server.ConnectionDescriptor.ConnectionState; -import io.moquette.server.netty.AutoFlushHandler; +import io.moquette.server.ConnectionDescriptorStore; import io.moquette.server.netty.NettyUtils; import io.moquette.spi.*; import io.moquette.spi.IMessagesStore.StoredMessage; @@ -120,7 +120,7 @@ public class ProtocolProcessor { private static final Logger LOG = LoggerFactory.getLogger(ProtocolProcessor.class); - protected ConcurrentMap connectionDescriptors; + protected ConnectionDescriptorStore connectionDescriptors; protected ConcurrentMap subscriptionInCourse; private SubscriptionsStore subscriptions; @@ -158,6 +158,15 @@ public class ProtocolProcessor { boolean allowZeroByteClientId, IAuthorizator authorizator, BrokerInterceptor interceptor) { init(subscriptions,storageService,sessionsStore,authenticator,allowAnonymous, allowZeroByteClientId, authorizator,interceptor,null); } + + public void init(SubscriptionsStore subscriptions, IMessagesStore storageService, + ISessionsStore sessionsStore, + IAuthenticator authenticator, + boolean allowAnonymous, + boolean allowZeroByteClientId, IAuthorizator authorizator, BrokerInterceptor interceptor, String serverPort) { + init(new ConnectionDescriptorStore(), subscriptions, storageService, sessionsStore, authenticator, + allowAnonymous, allowZeroByteClientId, authorizator, interceptor, serverPort); + } /** * @param subscriptions the subscription store where are stored all the existing @@ -171,12 +180,12 @@ public class ProtocolProcessor { * @param authorizator used to apply ACL policies to publishes and subscriptions. * @param interceptor to notify events to an intercept handler */ - void init(SubscriptionsStore subscriptions, IMessagesStore storageService, + void init(ConnectionDescriptorStore connectionDescriptors, SubscriptionsStore subscriptions, IMessagesStore storageService, ISessionsStore sessionsStore, IAuthenticator authenticator, boolean allowAnonymous, boolean allowZeroByteClientId, IAuthorizator authorizator, BrokerInterceptor interceptor, String serverPort) { - this.connectionDescriptors = new ConcurrentHashMap<>(); + this.connectionDescriptors = connectionDescriptors; this.subscriptionInCourse = new ConcurrentHashMap<>(); this.m_interceptor = interceptor; this.subscriptions = subscriptions; @@ -237,7 +246,7 @@ public class ProtocolProcessor { final String clientID = msg.getClientID(); ConnectionDescriptor descriptor = new ConnectionDescriptor(clientID, channel, msg.isCleanSession()); - ConnectionDescriptor existing = this.connectionDescriptors.putIfAbsent(clientID, descriptor); + ConnectionDescriptor existing = this.connectionDescriptors.addConnection(descriptor); if (existing != null) { LOG.info("Found an existing connection with same client ID <{}>, forcing to close", msg.getClientID()); existing.abort(); @@ -312,7 +321,7 @@ public class ProtocolProcessor { if (isSessionAlreadyStored) { clientSession.cleanSession(msg.isCleanSession()); } - descriptor.channel.writeAndFlush(okResp); + descriptor.writeAndFlush(okResp); return true; } @@ -373,7 +382,7 @@ public class ProtocolProcessor { republishStoredInSession(clientSession); } int flushIntervalMs = 500/*(keepAlive * 1000) / 2*/; - setupAutoFlusher(descriptor.channel.pipeline(), flushIntervalMs); + descriptor.setupAutoFlusher(flushIntervalMs); return true; } @@ -384,15 +393,6 @@ public class ProtocolProcessor { LOG.info("Client {} failed to connect with bad username or password.", session); } - private void setupAutoFlusher(ChannelPipeline pipeline, int flushIntervalMs) { - try { - pipeline.addAfter("idleEventHandler", "autoFlusher", new AutoFlushHandler(flushIntervalMs, TimeUnit.MILLISECONDS)); - } catch (NoSuchElementException nseex) { - //the idleEventHandler is not present on the pipeline - pipeline.addFirst("autoFlusher", new AutoFlushHandler(flushIntervalMs, TimeUnit.MILLISECONDS)); - } - } - private void setIdleTime(ChannelPipeline pipeline, int idleTime) { if (pipeline.names().contains("idleStateHandler")) { pipeline.remove("idleStateHandler"); @@ -576,39 +576,39 @@ public class ProtocolProcessor { public void processDisconnect(Channel channel) throws InterruptedException { channel.flush(); final String clientID = NettyUtils.clientID(channel); - final ConnectionDescriptor existingDescriptor = this.connectionDescriptors.get(clientID); + final ConnectionDescriptor existingDescriptor = this.connectionDescriptors.getConnection(clientID); if (existingDescriptor == null) { //another client with same ID removed the descriptor, we must exit channel.close(); return; } - if (existingDescriptor.channel != channel) { + if (existingDescriptor.doesNotUseChannel(channel)) { //another client saved it's descriptor, exit - channel.close(); + existingDescriptor.abort(); return; } if (!removeSubscriptions(existingDescriptor, clientID)) { - channel.close(); + existingDescriptor.abort(); return; } if (!dropStoredMessages(existingDescriptor, clientID)) { - channel.close(); + existingDescriptor.abort(); return; } if (!cleanWillMessageAndNotifyInterceptor(existingDescriptor, clientID)) { - channel.close(); + existingDescriptor.abort(); return; } - if (!closeChannel(existingDescriptor)) { + if (!existingDescriptor.close()) { return; } - boolean stillPresent = this.connectionDescriptors.remove(clientID, existingDescriptor); + boolean stillPresent = this.connectionDescriptors.removeConnection(existingDescriptor); if (!stillPresent) { //another descriptor was inserted return; @@ -653,23 +653,14 @@ public class ProtocolProcessor { //cleanup the will store m_willStore.remove(clientID); - String username = NettyUtils.userName(descriptor.channel); + String username = descriptor.getUsername(); m_interceptor.notifyClientDisconnected(clientID, username); return true; } - private boolean closeChannel(ConnectionDescriptor descriptor) { - final boolean success = descriptor.assignState(ConnectionState.INTERCEPTORS_NOTIFIED, ConnectionState.DISCONNECTED); - if (!success) { - return false; - } - descriptor.channel.close(); - return true; - } - public void processConnectionLost(String clientID, Channel channel) { ConnectionDescriptor oldConnDescr = new ConnectionDescriptor(clientID, channel, true); - connectionDescriptors.remove(clientID, oldConnDescr); + connectionDescriptors.removeConnection(oldConnDescr); //publish the Will message (if any) for the clientID if (m_willStore.containsKey(clientID)) { WillMessage will = m_willStore.get(clientID); 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 263b0a13..a458042b 100644 --- a/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java +++ b/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java @@ -2,7 +2,7 @@ package io.moquette.spi.impl; import io.moquette.parser.proto.messages.PubAckMessage; import io.moquette.parser.proto.messages.PublishMessage; -import io.moquette.server.ConnectionDescriptor; +import io.moquette.server.ConnectionDescriptorStore; import io.moquette.server.netty.NettyUtils; import io.moquette.spi.IMessagesStore; import io.moquette.spi.MessageGUID; @@ -14,7 +14,6 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.List; -import java.util.concurrent.ConcurrentMap; import static io.moquette.spi.impl.ProtocolProcessor.asStoredMessage; @@ -25,13 +24,13 @@ class Qos1PublishHandler { private final SubscriptionsStore subscriptions; private final IMessagesStore m_messagesStore; private final BrokerInterceptor m_interceptor; - private final ConcurrentMap connectionDescriptors; + private final ConnectionDescriptorStore connectionDescriptors; private final String brokerPort; private final MessagesPublisher publisher; public Qos1PublishHandler(IAuthorizator authorizator, SubscriptionsStore subscriptions, IMessagesStore messagesStore, BrokerInterceptor interceptor, - ConcurrentMap connectionDescriptors, + ConnectionDescriptorStore connectionDescriptors, String brokerPort, MessagesPublisher messagesPublisher) { this.m_authorizator = authorizator; this.subscriptions = subscriptions; @@ -103,10 +102,10 @@ class Qos1PublishHandler { throw new RuntimeException("Internal bad error, found connectionDescriptors to null while it should be initialized, somewhere it's overwritten!!"); } LOG.debug("clientIDs are {}", connectionDescriptors); - if (connectionDescriptors.get(clientId) == null) { + if (!connectionDescriptors.isConnected(clientId)) { throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client %s in cache %s", clientId, connectionDescriptors)); } - connectionDescriptors.get(clientId).channel.writeAndFlush(pubAckMessage); + connectionDescriptors.sendMessage(pubAckMessage, messageID, clientId); } 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 366e26db..5bc9651a 100644 --- a/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java +++ b/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java @@ -1,7 +1,7 @@ package io.moquette.spi.impl; import io.moquette.parser.proto.messages.*; -import io.moquette.server.ConnectionDescriptor; +import io.moquette.server.ConnectionDescriptorStore; import io.moquette.server.netty.NettyUtils; import io.moquette.spi.ClientSession; import io.moquette.spi.IMessagesStore; @@ -15,7 +15,6 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.List; -import java.util.concurrent.ConcurrentMap; import static io.moquette.spi.impl.ProtocolProcessor.asStoredMessage; @@ -27,14 +26,14 @@ class Qos2PublishHandler { private final SubscriptionsStore subscriptions; private final IMessagesStore m_messagesStore; private final BrokerInterceptor m_interceptor; - private final ConcurrentMap connectionDescriptors; + private final ConnectionDescriptorStore connectionDescriptors; private final ISessionsStore m_sessionsStore; private final String brokerPort; private final MessagesPublisher publisher; public Qos2PublishHandler(IAuthorizator authorizator, SubscriptionsStore subscriptions, IMessagesStore messagesStore, BrokerInterceptor interceptor, - ConcurrentMap connectionDescriptors, + ConnectionDescriptorStore connectionDescriptors, ISessionsStore sessionsStore, String brokerPort, MessagesPublisher messagesPublisher) { this.m_authorizator = authorizator; this.subscriptions = subscriptions; @@ -122,13 +121,13 @@ class Qos2PublishHandler { LOG.trace("PUB <--PUBREC-- SRV sendPubRec invoked for clientID {} with messageID {}", clientID, messageID); PubRecMessage pubRecMessage = new PubRecMessage(); pubRecMessage.setMessageID(messageID); - connectionDescriptors.get(clientID).channel.writeAndFlush(pubRecMessage); + connectionDescriptors.sendMessage(pubRecMessage,messageID, clientID); } private void sendPubComp(String clientID, int messageID) { LOG.debug("PUB <--PUBCOMP-- SRV sendPubComp invoked for clientID {} ad messageID {}", clientID, messageID); PubCompMessage pubCompMessage = new PubCompMessage(); pubCompMessage.setMessageID(messageID); - connectionDescriptors.get(clientID).channel.writeAndFlush(pubCompMessage); + connectionDescriptors.sendMessage(pubCompMessage, messageID, clientID); } }