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

Changed che willpublish to use the Qos2Publisher helper class

Browse Source
This commit is contained in:
andsel
2016-12-13 23:43:02 +01:00
parent a54b122094
commit 5b254cfe75
2 changed files with 11 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+8 -54
View File
@@ -136,6 +136,7 @@ public class ProtocolProcessor {
private Qos0PublishHandler qos0PublishHandler;
private Qos1PublishHandler qos1PublishHandler;
private Qos2PublishHandler qos2PublishHandler;
private Qos2Publisher qos2Publisher;
//maps clientID to Will testament, if specified on CONNECT
private ConcurrentMap<String, WillMessage> m_willStore = new ConcurrentHashMap<>();
@@ -191,8 +192,10 @@ public class ProtocolProcessor {
m_interceptor, this.connectionDescriptors, m_sessionsStore);
this.qos1PublishHandler = new Qos1PublishHandler(m_authorizator, subscriptions, m_messagesStore,
m_interceptor, this.connectionDescriptors, m_sessionsStore, m_server_port);
this.qos2Publisher = new Qos2Publisher(connectionDescriptors, sessionsStore, m_messagesStore);
this.qos2PublishHandler = new Qos2PublishHandler(m_authorizator, subscriptions, m_messagesStore,
m_interceptor, this.connectionDescriptors, m_sessionsStore, m_server_port);
m_interceptor, this.connectionDescriptors, m_sessionsStore, m_server_port, this.qos2Publisher);
}
public void processConnect(Channel channel, ConnectMessage msg) {
@@ -486,7 +489,8 @@ public class ProtocolProcessor {
guid = m_messagesStore.storePublishForFuture(toStoreMsg);
}
List<Subscription> topicMatchingSubscriptions = subscriptions.matches(topic);
publish2Subscribers(toStoreMsg, topicMatchingSubscriptions);
// publish2Subscribers(toStoreMsg, topicMatchingSubscriptions);
this.qos2Publisher.publish2Subscribers(toStoreMsg, topicMatchingSubscriptions);
if (!msg.isRetainFlag()) {
return;
@@ -519,60 +523,10 @@ public class ProtocolProcessor {
tobeStored.setMessageID(messageId);
String topic = tobeStored.getTopic();
List<Subscription> topicMatchingSubscriptions = subscriptions.matches(topic);
publish2Subscribers(tobeStored, topicMatchingSubscriptions);
this.qos2Publisher.publish2Subscribers(tobeStored, topicMatchingSubscriptions);
}
/**
* Flood the subscribers with the message to notify. MessageID is optional and should only used for QoS 1 and 2
* */
void publish2Subscribers(IMessagesStore.StoredMessage pubMsg, List<Subscription> topicMatchingSubscriptions) {
final String topic = pubMsg.getTopic();
final AbstractMessage.QOSType publishingQos = pubMsg.getQos();
final ByteBuffer origMessage = pubMsg.getMessage();
LOG.debug("publish2Subscribers republishing to existing subscribers that matches the topic {}", topic);
if (LOG.isTraceEnabled()) {
LOG.trace("content <{}>", DebugUtils.payload2Str(origMessage));
LOG.trace("subscription tree {}", subscriptions.dumpTree());
}
//if QoS 1 or 2 store the message
MessageGUID guid = null;
if (publishingQos != AbstractMessage.QOSType.MOST_ONE) {
guid = m_messagesStore.storePublishForFuture(pubMsg);
}
LOG.trace("Found {} matching subscriptions to <{}>", topicMatchingSubscriptions.size(), topic);
for (final Subscription sub : topicMatchingSubscriptions) {
AbstractMessage.QOSType qos = lowerQosToTheSubscriptionDesired(sub, publishingQos);
ClientSession targetSession = m_sessionsStore.sessionForClient(sub.getClientId());
boolean targetIsActive = this.connectionDescriptors.containsKey(sub.getClientId());
LOG.debug("Broker republishing to client <{}> topic <{}> qos <{}>, active {}",
sub.getClientId(), sub.getTopicFilter(), qos, targetIsActive);
ByteBuffer message = origMessage.duplicate();
if (qos == AbstractMessage.QOSType.MOST_ONE && targetIsActive) {
//QoS 0
directSend(targetSession, topic, qos, message, false, null);
} else {
//QoS 1 or 2
//if the target subscription is not clean session and is not connected => store it
if (!targetSession.isCleanSession() && !targetIsActive) {
//store the message in targetSession queue to deliver
targetSession.enqueueToDeliver(guid);
} else {
//publish
if (targetIsActive) {
int messageId = targetSession.nextPacketId();
targetSession.inFlightAckWaiting(guid, messageId);
directSend(targetSession, topic, qos, message, false, messageId);
}
}
}
}
}
static QOSType lowerQosToTheSubscriptionDesired(Subscription sub, QOSType qos) {
if (qos.byteValue() > sub.getRequestedQos().byteValue()) {
qos = sub.getRequestedQos();
broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java
+3 -2
View File
@@ -34,7 +34,8 @@ class Qos2PublishHandler {
public Qos2PublishHandler(IAuthorizator authorizator, SubscriptionsStore subscriptions,
IMessagesStore messagesStore, BrokerInterceptor interceptor, ConcurrentMap<String,
ConnectionDescriptor> connectionDescriptors, ISessionsStore sessionsStore, String brokerPort) {
ConnectionDescriptor> connectionDescriptors, ISessionsStore sessionsStore, String brokerPort,
Qos2Publisher qos2Publisher) {
this.m_authorizator = authorizator;
this.subscriptions = subscriptions;
this.m_messagesStore = messagesStore;
@@ -42,7 +43,7 @@ class Qos2PublishHandler {
this.connectionDescriptors = connectionDescriptors;
this.m_sessionsStore = sessionsStore;
this.brokerPort = brokerPort;
this.qos2Publisher = new Qos2Publisher(connectionDescriptors, sessionsStore, messagesStore);
this.qos2Publisher = qos2Publisher;
}
void receivedPublishQos2(Channel channel, PublishMessage msg) {