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 e3a4add3..ec3bbac0 100644 --- a/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java +++ b/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java @@ -3,7 +3,6 @@ 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.spi.ClientSession; import io.moquette.spi.IMessagesStore; import org.slf4j.Logger; @@ -12,18 +11,15 @@ import org.slf4j.LoggerFactory; import java.nio.ByteBuffer; import java.util.Collection; import java.util.List; -import java.util.concurrent.ConcurrentMap; class InternalRepublisher { private static final Logger LOG = LoggerFactory.getLogger(InternalRepublisher.class); - private final ConcurrentMap connectionDescriptors; - private final PersistentQueueMessageSender persistentSender; + private final PersistentQueueMessageSender messageSender; - InternalRepublisher(ConcurrentMap connectionDescriptors) { - this.connectionDescriptors = connectionDescriptors; - this.persistentSender = new PersistentQueueMessageSender(connectionDescriptors); + InternalRepublisher(PersistentQueueMessageSender messageSender) { + this.messageSender = messageSender; } void publishRetained(ClientSession targetSession, Collection messages) { @@ -39,7 +35,7 @@ class InternalRepublisher { if (publishMsg.getQos() != AbstractMessage.QOSType.MOST_ONE) { publishMsg.setMessageID(packetID); } - this.persistentSender.sendPublish(targetSession, publishMsg); + this.messageSender.sendPublish(targetSession, publishMsg); } } @@ -53,7 +49,7 @@ class InternalRepublisher { if (publishMsg.getQos() != AbstractMessage.QOSType.MOST_ONE) { publishMsg.setMessageID(pubEvt.getMessageID()); } - this.persistentSender.sendPublish(clientSession, publishMsg); + this.messageSender.sendPublish(clientSession, publishMsg); } } 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 7e983e2d..1c4146a4 100644 --- a/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java +++ b/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java @@ -23,17 +23,17 @@ class MessagesPublisher { private final ConcurrentMap connectionDescriptors; private final ISessionsStore m_sessionsStore; private final IMessagesStore m_messagesStore; - private final PersistentQueueMessageSender persistentSender; + private final PersistentQueueMessageSender messageSender; public MessagesPublisher(ConcurrentMap connectionDescriptors, ISessionsStore sessionsStore, - IMessagesStore messagesStore) { + IMessagesStore messagesStore, PersistentQueueMessageSender messageSender) { this.connectionDescriptors = connectionDescriptors; this.m_sessionsStore = sessionsStore; this.m_messagesStore = messagesStore; - this.persistentSender = new PersistentQueueMessageSender(connectionDescriptors); + this.messageSender = messageSender; } - static PublishMessage notRetainedPublishForQos(String topic, AbstractMessage.QOSType qos, ByteBuffer message) { + static PublishMessage notRetainedPublish(String topic, AbstractMessage.QOSType qos, ByteBuffer message) { PublishMessage pubMessage = new PublishMessage(); pubMessage.setRetainFlag(false); pubMessage.setTopicName(topic); @@ -64,7 +64,7 @@ class MessagesPublisher { sub.getClientId(), sub.getTopicFilter(), qos, targetIsActive); ByteBuffer message = origMessage.duplicate(); if (targetIsActive) { - PublishMessage publishMsg = notRetainedPublishForQos(topic, qos, message); + PublishMessage publishMsg = notRetainedPublish(topic, qos, message); if (qos != AbstractMessage.QOSType.MOST_ONE) { //QoS 1 or 2 int messageId = targetSession.nextPacketId(); @@ -72,7 +72,7 @@ class MessagesPublisher { //set the PacketIdentifier only for QoS > 0 publishMsg.setMessageID(messageId); } - this.persistentSender.sendPublish(targetSession, publishMsg); + this.messageSender.sendPublish(targetSession, publishMsg); } else { if (!targetSession.isCleanSession()) { //store the message in targetSession queue to deliver 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 55665195..ee4d7ee4 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -189,16 +189,17 @@ public class ProtocolProcessor { m_sessionsStore = sessionsStore; m_server_port = serverPort; + final PersistentQueueMessageSender messageSender = new PersistentQueueMessageSender(this.connectionDescriptors); + this.messagesPublisher = new MessagesPublisher(connectionDescriptors, sessionsStore, m_messagesStore, messageSender); + this.qos0PublishHandler = new Qos0PublishHandler(m_authorizator, subscriptions, m_messagesStore, - m_interceptor, this.connectionDescriptors, m_sessionsStore); + m_interceptor, this.messagesPublisher); this.qos1PublishHandler = new Qos1PublishHandler(m_authorizator, subscriptions, m_messagesStore, - m_interceptor, this.connectionDescriptors, m_sessionsStore, m_server_port); - - this.messagesPublisher = new MessagesPublisher(connectionDescriptors, sessionsStore, m_messagesStore); + m_interceptor, this.connectionDescriptors, m_server_port, this.messagesPublisher); this.qos2PublishHandler = new Qos2PublishHandler(m_authorizator, subscriptions, m_messagesStore, - m_interceptor, this.connectionDescriptors, m_sessionsStore, m_server_port/*, this.qos2Publisher*/); + m_interceptor, this.connectionDescriptors, m_sessionsStore, m_server_port, this.messagesPublisher); - this.internalRepublisher = new InternalRepublisher(this.connectionDescriptors); + this.internalRepublisher = new InternalRepublisher(messageSender); } public void processConnect(Channel channel, ConnectMessage msg) { diff --git a/broker/src/main/java/io/moquette/spi/impl/Qos0PublishHandler.java b/broker/src/main/java/io/moquette/spi/impl/Qos0PublishHandler.java index 559f0a0b..0adeebd9 100644 --- a/broker/src/main/java/io/moquette/spi/impl/Qos0PublishHandler.java +++ b/broker/src/main/java/io/moquette/spi/impl/Qos0PublishHandler.java @@ -1,10 +1,8 @@ package io.moquette.spi.impl; import io.moquette.parser.proto.messages.PublishMessage; -import io.moquette.server.ConnectionDescriptor; import io.moquette.server.netty.NettyUtils; import io.moquette.spi.IMessagesStore; -import io.moquette.spi.ISessionsStore; import io.moquette.spi.impl.subscriptions.Subscription; import io.moquette.spi.impl.subscriptions.SubscriptionsStore; import io.moquette.spi.security.IAuthorizator; @@ -13,7 +11,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,16 +22,16 @@ class Qos0PublishHandler { private final SubscriptionsStore subscriptions; private final IMessagesStore m_messagesStore; private final BrokerInterceptor m_interceptor; - private final MessagesPublisher metaPublisher; + private final MessagesPublisher publisher; public Qos0PublishHandler(IAuthorizator authorizator, SubscriptionsStore subscriptions, - IMessagesStore messagesStore, BrokerInterceptor interceptor, ConcurrentMap connectionDescriptors, ISessionsStore sessionsStore) { + IMessagesStore messagesStore, BrokerInterceptor interceptor, + MessagesPublisher messagesPublisher) { this.m_authorizator = authorizator; this.subscriptions = subscriptions; this.m_messagesStore = messagesStore; this.m_interceptor = interceptor; - this.metaPublisher = new MessagesPublisher(connectionDescriptors, sessionsStore, messagesStore); + this.publisher = messagesPublisher; } void receivedPublishQos0(Channel channel, PublishMessage msg) { @@ -55,7 +52,7 @@ class Qos0PublishHandler { LOG.trace("subscription tree {}", subscriptions.dumpTree()); } List topicMatchingSubscriptions = subscriptions.matches(topic); - this.metaPublisher.publish2Subscribers(toStoreMsg, topicMatchingSubscriptions); + this.publisher.publish2Subscribers(toStoreMsg, topicMatchingSubscriptions); if (msg.isRetainFlag()) { //QoS == 0 && retain => clean old retained 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 6060a0c3..263b0a13 100644 --- a/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java +++ b/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java @@ -5,7 +5,6 @@ import io.moquette.parser.proto.messages.PublishMessage; import io.moquette.server.ConnectionDescriptor; import io.moquette.server.netty.NettyUtils; import io.moquette.spi.IMessagesStore; -import io.moquette.spi.ISessionsStore; import io.moquette.spi.MessageGUID; import io.moquette.spi.impl.subscriptions.Subscription; import io.moquette.spi.impl.subscriptions.SubscriptionsStore; @@ -28,18 +27,19 @@ class Qos1PublishHandler { private final BrokerInterceptor m_interceptor; private final ConcurrentMap connectionDescriptors; private final String brokerPort; - private final MessagesPublisher metaPublisher; + private final MessagesPublisher publisher; public Qos1PublishHandler(IAuthorizator authorizator, SubscriptionsStore subscriptions, - IMessagesStore messagesStore, BrokerInterceptor interceptor, ConcurrentMap connectionDescriptors, ISessionsStore sessionsStore, String brokerPort) { + IMessagesStore messagesStore, BrokerInterceptor interceptor, + ConcurrentMap connectionDescriptors, + String brokerPort, MessagesPublisher messagesPublisher) { this.m_authorizator = authorizator; this.subscriptions = subscriptions; this.m_messagesStore = messagesStore; this.m_interceptor = interceptor; this.connectionDescriptors = connectionDescriptors; this.brokerPort = brokerPort; - this.metaPublisher = new MessagesPublisher(connectionDescriptors, sessionsStore, messagesStore); + this.publisher = messagesPublisher; } void receivedPublishQos1(Channel channel, PublishMessage msg) { @@ -60,7 +60,7 @@ class Qos1PublishHandler { LOG.trace("subscription tree {}", subscriptions.dumpTree()); } List topicMatchingSubscriptions = subscriptions.matches(topic); - this.metaPublisher.publish2Subscribers(toStoreMsg, topicMatchingSubscriptions); + this.publisher.publish2Subscribers(toStoreMsg, topicMatchingSubscriptions); //send PUBACK final Integer messageID = msg.getMessageID(); 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 50f67961..366e26db 100644 --- a/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java +++ b/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java @@ -30,12 +30,12 @@ class Qos2PublishHandler { private final ConcurrentMap connectionDescriptors; private final ISessionsStore m_sessionsStore; private final String brokerPort; - private final MessagesPublisher metaPublisher; + private final MessagesPublisher publisher; public Qos2PublishHandler(IAuthorizator authorizator, SubscriptionsStore subscriptions, - IMessagesStore messagesStore, BrokerInterceptor interceptor, ConcurrentMap connectionDescriptors, ISessionsStore sessionsStore, String brokerPort/*, - PersistentQueueMessageSender qos2Publisher*/) { + IMessagesStore messagesStore, BrokerInterceptor interceptor, + ConcurrentMap connectionDescriptors, + ISessionsStore sessionsStore, String brokerPort, MessagesPublisher messagesPublisher) { this.m_authorizator = authorizator; this.subscriptions = subscriptions; this.m_messagesStore = messagesStore; @@ -43,7 +43,7 @@ class Qos2PublishHandler { this.connectionDescriptors = connectionDescriptors; this.m_sessionsStore = sessionsStore; this.brokerPort = brokerPort; - this.metaPublisher = new MessagesPublisher(connectionDescriptors, sessionsStore, messagesStore); + this.publisher = messagesPublisher; } void receivedPublishQos2(Channel channel, PublishMessage msg) { @@ -95,7 +95,7 @@ class Qos2PublishHandler { LOG.trace("content <{}>", DebugUtils.payload2Str(evt.getMessage())); LOG.trace("subscription tree {}", subscriptions.dumpTree()); } - this.metaPublisher.publish2Subscribers(evt, topicMatchingSubscriptions); + this.publisher.publish2Subscribers(evt, topicMatchingSubscriptions); if (evt.isRetained()) { if (!evt.getMessage().hasRemaining()) {