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

Introduced 2 method, one for retained and not

Browse Source
This commit is contained in:
andsel
2016-12-13 23:43:02 +01:00
parent 633365672c
commit 57586ce7c0
3 changed files with 20 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java
+10 -2
View File
@@ -37,7 +37,7 @@ class InternalRepublisher {
LOG.trace("Adding to inflight <{}>", packetID);
targetSession.inFlightAckWaiting(storedMsg.getGuid(), packetID);
}
PublishMessage publishMsg = createPublishForQos(storedMsg.getTopic(), storedMsg.getQos(), storedMsg.getPayload(), true);
PublishMessage publishMsg = retainedPublishForQos(storedMsg.getTopic(), storedMsg.getQos(), storedMsg.getPayload());
//set the PacketIdentifier only for QoS > 0
if (publishMsg.getQos() != AbstractMessage.QOSType.MOST_ONE) {
publishMsg.setMessageID(packetID);
@@ -51,7 +51,7 @@ class InternalRepublisher {
//put in flight zone
LOG.trace("Adding to inflight <{}>", pubEvt.getMessageID());
clientSession.inFlightAckWaiting(pubEvt.getGuid(), pubEvt.getMessageID());
PublishMessage publishMsg = createPublishForQos(pubEvt.getTopic(), pubEvt.getQos(), pubEvt.getMessage(), false);
PublishMessage publishMsg = notRetainedPublishForQos(pubEvt.getTopic(), pubEvt.getQos(), pubEvt.getMessage());
//set the PacketIdentifier only for QoS > 0
if (publishMsg.getQos() != AbstractMessage.QOSType.MOST_ONE) {
publishMsg.setMessageID(pubEvt.getMessageID());
@@ -60,6 +60,14 @@ class InternalRepublisher {
}
}
private PublishMessage notRetainedPublishForQos(String topic, AbstractMessage.QOSType qos, ByteBuffer message) {
return createPublishForQos(topic, qos, message, false);
}
private PublishMessage retainedPublishForQos(String topic, AbstractMessage.QOSType qos, ByteBuffer message) {
return createPublishForQos(topic, qos, message, true);
}
private PublishMessage createPublishForQos(String topic, AbstractMessage.QOSType qos, ByteBuffer message, boolean retained) {
PublishMessage pubMessage = new PublishMessage();
pubMessage.setRetainFlag(retained);
broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java
+10 -2
View File
@@ -15,7 +15,6 @@ import java.nio.ByteBuffer;
import java.util.List;
import java.util.concurrent.ConcurrentMap;
import static io.moquette.spi.impl.PersistentQueueMessageSender.createPublishForQos;
import static io.moquette.spi.impl.ProtocolProcessor.lowerQosToTheSubscriptionDesired;
class MessagesPublisher {
@@ -34,6 +33,15 @@ class MessagesPublisher {
this.persistentSender = new PersistentQueueMessageSender(connectionDescriptors);
}
static PublishMessage notRetainedPublishForQos(String topic, AbstractMessage.QOSType qos, ByteBuffer message) {
PublishMessage pubMessage = new PublishMessage();
pubMessage.setRetainFlag(false);
pubMessage.setTopicName(topic);
pubMessage.setQos(qos);
pubMessage.setPayload(message);
return pubMessage;
}
void publish2Subscribers(IMessagesStore.StoredMessage pubMsg, List<Subscription> topicMatchingSubscriptions) {
final String topic = pubMsg.getTopic();
final AbstractMessage.QOSType publishingQos = pubMsg.getQos();
@@ -56,7 +64,7 @@ class MessagesPublisher {
sub.getClientId(), sub.getTopicFilter(), qos, targetIsActive);
ByteBuffer message = origMessage.duplicate();
if (targetIsActive) {
PublishMessage publishMsg = createPublishForQos(topic, qos, message);
PublishMessage publishMsg = notRetainedPublishForQos(topic, qos, message);
if (qos != AbstractMessage.QOSType.MOST_ONE) {
//QoS 1 or 2
int messageId = targetSession.nextPacketId();
broker/src/main/java/io/moquette/spi/impl/PersistentQueueMessageSender.java
-12
View File
@@ -1,6 +1,5 @@
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;
@@ -8,7 +7,6 @@ import io.netty.channel.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.nio.ByteBuffer;
import java.util.concurrent.ConcurrentMap;
import static io.moquette.parser.proto.messages.AbstractMessage.QOSType.MOST_ONE;
@@ -53,14 +51,4 @@ class PersistentQueueMessageSender {
clientsession.enqueue(pubMessage);
}
}
static PublishMessage createPublishForQos(String topic, AbstractMessage.QOSType qos, ByteBuffer message) {
PublishMessage pubMessage = new PublishMessage();
pubMessage.setRetainFlag(false);
pubMessage.setTopicName(topic);
pubMessage.setQos(qos);
pubMessage.setPayload(message);
return pubMessage;
}
}