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

Reinserted route to qo0 branch in publish2Subscribers for internal publish

Browse Source
This commit is contained in:
andsel
2016-12-13 23:43:02 +01:00
parent 12ed805737
commit 81a447dce9
1 changed files with 16 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+16 -11
View File
@@ -622,17 +622,22 @@ public class ProtocolProcessor {
LOG.debug("Broker republishing to client <{}> topic <{}> qos <{}>, active {}",
sub.getClientId(), sub.getTopicFilter(), qos, targetIsActive);
ByteBuffer message = origMessage.duplicate();
//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);
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);
}
}
}
}