This commit is contained in:
andsel
2016-12-13 23:43:02 +01:00
parent c76db85ae7
commit 0e0afe7c9b
4 changed files with 13 additions and 13 deletions
@@ -11,12 +11,12 @@ import org.slf4j.LoggerFactory;
import java.nio.ByteBuffer;
import java.util.concurrent.ConcurrentMap;
class Qos0Publisher {
class BestEffortMessageSender {
private static final Logger LOG = LoggerFactory.getLogger(Qos0Publisher.class);
private static final Logger LOG = LoggerFactory.getLogger(BestEffortMessageSender.class);
private final ConcurrentMap<String, ConnectionDescriptor> connectionDescriptors;
public Qos0Publisher(ConcurrentMap<String, ConnectionDescriptor> connectionDescriptors) {
public BestEffortMessageSender(ConcurrentMap<String, ConnectionDescriptor> connectionDescriptors) {
this.connectionDescriptors = connectionDescriptors;
}
@@ -22,16 +22,16 @@ class MetaPublisher {
private final ConcurrentMap<String, ConnectionDescriptor> connectionDescriptors;
private final ISessionsStore m_sessionsStore;
private final IMessagesStore m_messagesStore;
private final Qos0Publisher qos0Publisher;
private final Qos2Publisher qos2Publisher;
private final BestEffortMessageSender bestEffortSender;
private final PersistentQueueMessageSender persistentSender;
public MetaPublisher(ConcurrentMap<String, ConnectionDescriptor> connectionDescriptors, ISessionsStore sessionsStore,
IMessagesStore messagesStore) {
this.connectionDescriptors = connectionDescriptors;
this.m_sessionsStore = sessionsStore;
this.m_messagesStore = messagesStore;
this.qos0Publisher = new Qos0Publisher(connectionDescriptors);
this.qos2Publisher = new Qos2Publisher(connectionDescriptors);
this.bestEffortSender = new BestEffortMessageSender(connectionDescriptors);
this.persistentSender = new PersistentQueueMessageSender(connectionDescriptors);
}
void publish2Subscribers(IMessagesStore.StoredMessage pubMsg, List<Subscription> topicMatchingSubscriptions) {
@@ -58,12 +58,12 @@ class MetaPublisher {
if (targetIsActive) {
//QoS 0
if (qos == AbstractMessage.QOSType.MOST_ONE) {
this.qos0Publisher.publishQos0(targetSession, topic, message);
this.bestEffortSender.publishQos0(targetSession, topic, message);
} else {
//QoS 1 or 2
int messageId = targetSession.nextPacketId();
targetSession.inFlightAckWaiting(guid, messageId);
this.qos2Publisher.publishQos2(targetSession, topic, qos, message, false, messageId);
this.persistentSender.publishQos2(targetSession, topic, qos, message, false, messageId);
}
} else {
if (!targetSession.isCleanSession()) {
@@ -11,12 +11,12 @@ import org.slf4j.LoggerFactory;
import java.nio.ByteBuffer;
import java.util.concurrent.ConcurrentMap;
class Qos2Publisher {
class PersistentQueueMessageSender {
private static final Logger LOG = LoggerFactory.getLogger(Qos2Publisher.class);
private static final Logger LOG = LoggerFactory.getLogger(PersistentQueueMessageSender.class);
private final ConcurrentMap<String, ConnectionDescriptor> connectionDescriptors;
public Qos2Publisher(ConcurrentMap<String, ConnectionDescriptor> connectionDescriptors) {
public PersistentQueueMessageSender(ConcurrentMap<String, ConnectionDescriptor> connectionDescriptors) {
this.connectionDescriptors = connectionDescriptors;
}
@@ -35,7 +35,7 @@ class Qos2PublishHandler {
public Qos2PublishHandler(IAuthorizator authorizator, SubscriptionsStore subscriptions,
IMessagesStore messagesStore, BrokerInterceptor interceptor, ConcurrentMap<String,
ConnectionDescriptor> connectionDescriptors, ISessionsStore sessionsStore, String brokerPort/*,
Qos2Publisher qos2Publisher*/) {
PersistentQueueMessageSender qos2Publisher*/) {
this.m_authorizator = authorizator;
this.subscriptions = subscriptions;
this.m_messagesStore = messagesStore;