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

Condensed the looping on subscription on separate class

Browse Source
This commit is contained in:
andsel
2016-12-13 23:43:02 +01:00
parent b58cd086ea
commit 2b554ca475
8 changed files with 96 additions and 208 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/spi/impl/Qos2Publisher.java
+1 -47
View File
@@ -6,18 +6,13 @@ import io.moquette.server.ConnectionDescriptor;
import io.moquette.spi.ClientSession;
import io.moquette.spi.IMessagesStore;
import io.moquette.spi.ISessionsStore;
import io.moquette.spi.MessageGUID;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.netty.channel.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.concurrent.ConcurrentMap;
import static io.moquette.spi.impl.ProtocolProcessor.lowerQosToTheSubscriptionDesired;
class Qos2Publisher {
private static final Logger LOG = LoggerFactory.getLogger(Qos2Publisher.class);
@@ -32,48 +27,7 @@ class Qos2Publisher {
m_messagesStore = messagesStore;
}
void publish2Subscribers(IMessagesStore.StoredMessage pubMsg, List<Subscription> topicMatchingSubscriptions) {
final String topic = pubMsg.getTopic();
final AbstractMessage.QOSType publishingQos = pubMsg.getQos();
final ByteBuffer origMessage = pubMsg.getMessage();
//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 (targetIsActive) {
//QoS 0
if (qos == AbstractMessage.QOSType.MOST_ONE) {
publishQos2(targetSession, topic, qos, message, false, null);
} else {
//QoS 1 or 2
int messageId = targetSession.nextPacketId();
targetSession.inFlightAckWaiting(guid, messageId);
publishQos2(targetSession, topic, qos, message, false, messageId);
}
} else {
if (!targetSession.isCleanSession()) {
//store the message in targetSession queue to deliver
targetSession.enqueueToDeliver(guid);
}
}
}
}
private void publishQos2(ClientSession clientsession, String topic, AbstractMessage.QOSType qos,
void publishQos2(ClientSession clientsession, String topic, AbstractMessage.QOSType qos,
ByteBuffer message, boolean retained, Integer messageID) {
String clientId = clientsession.clientID;
LOG.debug("directSend invoked clientId <{}> on topic <{}> QoS {} retained {} messageID {}",