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

Removed data clump

Browse Source
This commit is contained in:
andsel
2016-12-13 23:43:02 +01:00
parent 57586ce7c0
commit f7e8da93d8
1 changed files with 6 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java
+6 -9
View File
@@ -6,7 +6,6 @@ import io.moquette.parser.proto.messages.PublishMessage;
import io.moquette.server.ConnectionDescriptor;
import io.moquette.spi.ClientSession;
import io.moquette.spi.IMessagesStore;
import io.netty.channel.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -15,8 +14,6 @@ import java.util.Collection;
import java.util.List;
import java.util.concurrent.ConcurrentMap;
import static io.moquette.parser.proto.messages.AbstractMessage.QOSType.MOST_ONE;
class InternalRepublisher {
private static final Logger LOG = LoggerFactory.getLogger(InternalRepublisher.class);
@@ -37,7 +34,7 @@ class InternalRepublisher {
LOG.trace("Adding to inflight <{}>", packetID);
targetSession.inFlightAckWaiting(storedMsg.getGuid(), packetID);
}
PublishMessage publishMsg = retainedPublishForQos(storedMsg.getTopic(), storedMsg.getQos(), storedMsg.getPayload());
PublishMessage publishMsg = retainedPublish(storedMsg);
//set the PacketIdentifier only for QoS > 0
if (publishMsg.getQos() != AbstractMessage.QOSType.MOST_ONE) {
publishMsg.setMessageID(packetID);
@@ -51,7 +48,7 @@ class InternalRepublisher {
//put in flight zone
LOG.trace("Adding to inflight <{}>", pubEvt.getMessageID());
clientSession.inFlightAckWaiting(pubEvt.getGuid(), pubEvt.getMessageID());
PublishMessage publishMsg = notRetainedPublishForQos(pubEvt.getTopic(), pubEvt.getQos(), pubEvt.getMessage());
PublishMessage publishMsg = notRetainedPublish(pubEvt);
//set the PacketIdentifier only for QoS > 0
if (publishMsg.getQos() != AbstractMessage.QOSType.MOST_ONE) {
publishMsg.setMessageID(pubEvt.getMessageID());
@@ -60,12 +57,12 @@ class InternalRepublisher {
}
}
private PublishMessage notRetainedPublishForQos(String topic, AbstractMessage.QOSType qos, ByteBuffer message) {
return createPublishForQos(topic, qos, message, false);
private PublishMessage notRetainedPublish(IMessagesStore.StoredMessage storedMessage) {
return createPublishForQos(storedMessage.getTopic(), storedMessage.getQos(), storedMessage.getMessage(), false);
}
private PublishMessage retainedPublishForQos(String topic, AbstractMessage.QOSType qos, ByteBuffer message) {
return createPublishForQos(topic, qos, message, true);
private PublishMessage retainedPublish(IMessagesStore.StoredMessage storedMessage) {
return createPublishForQos(storedMessage.getTopic(), storedMessage.getQos(), storedMessage.getMessage(), true);
}
private PublishMessage createPublishForQos(String topic, AbstractMessage.QOSType qos, ByteBuffer message, boolean retained) {