anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Flowed up the creation of messageSender

Browse Source
This commit is contained in:
andsel
2016-12-13 23:43:02 +01:00
parent f7e8da93d8
commit 1831560bd6
6 changed files with 35 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java
+5 -9
View File
@@ -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<String, ConnectionDescriptor> connectionDescriptors;
private final PersistentQueueMessageSender persistentSender;
private final PersistentQueueMessageSender messageSender;
InternalRepublisher(ConcurrentMap<String, ConnectionDescriptor> connectionDescriptors) {
this.connectionDescriptors = connectionDescriptors;
this.persistentSender = new PersistentQueueMessageSender(connectionDescriptors);
InternalRepublisher(PersistentQueueMessageSender messageSender) {
this.messageSender = messageSender;
}
void publishRetained(ClientSession targetSession, Collection<IMessagesStore.StoredMessage> 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);
}
}
broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java
+6 -6
View File
@@ -23,17 +23,17 @@ class MessagesPublisher {
private final ConcurrentMap<String, ConnectionDescriptor> connectionDescriptors;
private final ISessionsStore m_sessionsStore;
private final IMessagesStore m_messagesStore;
private final PersistentQueueMessageSender persistentSender;
private final PersistentQueueMessageSender messageSender;
public MessagesPublisher(ConcurrentMap<String, ConnectionDescriptor> 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
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+7 -6
View File
@@ -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) {
broker/src/main/java/io/moquette/spi/impl/Qos0PublishHandler.java
+5 -8
View File
@@ -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<String,
ConnectionDescriptor> 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<Subscription> topicMatchingSubscriptions = subscriptions.matches(topic);
this.metaPublisher.publish2Subscribers(toStoreMsg, topicMatchingSubscriptions);
this.publisher.publish2Subscribers(toStoreMsg, topicMatchingSubscriptions);
if (msg.isRetainFlag()) {
//QoS == 0 && retain => clean old retained
broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java
+6 -6
View File
@@ -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<String, ConnectionDescriptor> connectionDescriptors;
private final String brokerPort;
private final MessagesPublisher metaPublisher;
private final MessagesPublisher publisher;
public Qos1PublishHandler(IAuthorizator authorizator, SubscriptionsStore subscriptions,
IMessagesStore messagesStore, BrokerInterceptor interceptor, ConcurrentMap<String,
ConnectionDescriptor> connectionDescriptors, ISessionsStore sessionsStore, String brokerPort) {
IMessagesStore messagesStore, BrokerInterceptor interceptor,
ConcurrentMap<String, ConnectionDescriptor> 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<Subscription> topicMatchingSubscriptions = subscriptions.matches(topic);
this.metaPublisher.publish2Subscribers(toStoreMsg, topicMatchingSubscriptions);
this.publisher.publish2Subscribers(toStoreMsg, topicMatchingSubscriptions);
//send PUBACK
final Integer messageID = msg.getMessageID();
broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java
+6 -6
View File
@@ -30,12 +30,12 @@ class Qos2PublishHandler {
private final ConcurrentMap<String, ConnectionDescriptor> 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<String,
ConnectionDescriptor> connectionDescriptors, ISessionsStore sessionsStore, String brokerPort/*,
PersistentQueueMessageSender qos2Publisher*/) {
IMessagesStore messagesStore, BrokerInterceptor interceptor,
ConcurrentMap<String, ConnectionDescriptor> 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()) {