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

Separated qos0 publishing logic from publish handler

Browse Source
This commit is contained in:
andsel
2016-12-13 23:43:02 +01:00
parent e6bb883334
commit a53772f094
2 changed files with 96 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/spi/impl/Qos0PublishHandler.java
+8 -67
View File
@@ -1,10 +1,8 @@
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.server.netty.NettyUtils;
import io.moquette.spi.ClientSession;
import io.moquette.spi.IMessagesStore;
import io.moquette.spi.ISessionsStore;
import io.moquette.spi.impl.subscriptions.Subscription;
@@ -14,7 +12,6 @@ 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;
@@ -28,8 +25,7 @@ class Qos0PublishHandler {
private final SubscriptionsStore subscriptions;
private final IMessagesStore m_messagesStore;
private final BrokerInterceptor m_interceptor;
private final ConcurrentMap<String, ConnectionDescriptor> connectionDescriptors;
private final ISessionsStore m_sessionsStore;
private final Qos0Publisher qos0Publisher;
public Qos0PublishHandler(IAuthorizator authorizator, SubscriptionsStore subscriptions,
IMessagesStore messagesStore, BrokerInterceptor interceptor, ConcurrentMap<String,
@@ -38,8 +34,7 @@ class Qos0PublishHandler {
this.subscriptions = subscriptions;
this.m_messagesStore = messagesStore;
this.m_interceptor = interceptor;
this.connectionDescriptors = connectionDescriptors;
this.m_sessionsStore = sessionsStore;
this.qos0Publisher = new Qos0Publisher(connectionDescriptors, sessionsStore);
}
void receivedPublishQos0(Channel channel, PublishMessage msg) {
@@ -54,8 +49,13 @@ class Qos0PublishHandler {
String clientID = NettyUtils.clientID(channel);
toStoreMsg.setClientID(clientID);
LOG.debug("publish2Subscribers republishing to existing subscribers that matches the topic {}", topic);
if (LOG.isTraceEnabled()) {
LOG.trace("content <{}>", DebugUtils.payload2Str(toStoreMsg.getMessage()));
LOG.trace("subscription tree {}", subscriptions.dumpTree());
}
List<Subscription> topicMatchingSubscriptions = subscriptions.matches(topic);
publish2Subscribers_qos0(toStoreMsg, topicMatchingSubscriptions);
qos0Publisher.publish2Subscribers_qos0(toStoreMsg, topicMatchingSubscriptions);
if (msg.isRetainFlag()) {
//QoS == 0 && retain => clean old retained
@@ -75,63 +75,4 @@ class Qos0PublishHandler {
}
return false;
}
private void publish2Subscribers_qos0(IMessagesStore.StoredMessage pubMsg, List<Subscription> topicMatchingSubscriptions) {
final String topic = pubMsg.getTopic();
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());
}
LOG.trace("Found {} matching subscriptions to <{}>", topicMatchingSubscriptions.size(), topic);
for (final Subscription sub : topicMatchingSubscriptions) {
boolean targetIsActive = connectionDescriptors.containsKey(sub.getClientId());
LOG.debug("Broker republishing to client <{}> topic <{}> qos <{}>, active {}",
sub.getClientId(), sub.getTopicFilter(), AbstractMessage.QOSType.MOST_ONE, targetIsActive);
if (targetIsActive) {
ClientSession targetSession = m_sessionsStore.sessionForClient(sub.getClientId());
ByteBuffer message = origMessage.duplicate();
publishQos0(targetSession, topic, message);
}
}
}
protected void publishQos0(ClientSession clientsession, String topic, ByteBuffer message) {
String clientId = clientsession.clientID;
LOG.debug("publishQos0 invoked clientId <{}> on topic <{}>", clientId, topic);
PublishMessage pubMessage = new PublishMessage();
pubMessage.setRetainFlag(false);
pubMessage.setTopicName(topic);
pubMessage.setQos(AbstractMessage.QOSType.MOST_ONE);
pubMessage.setPayload(message);
LOG.info("send publish message to <{}> on topic <{}>", clientId, topic);
if (LOG.isDebugEnabled()) {
LOG.debug("content <{}>", DebugUtils.payload2Str(message));
}
if (connectionDescriptors == null) {
throw new RuntimeException("Internal bad error, found connectionDescriptors to null while it should be " +
"initialized, somewhere it's overwritten!!");
}
if (connectionDescriptors.get(clientId) == null) {
//TODO while we were publishing to the target client, that client disconnected,
// could happen is not an error HANDLE IT
throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client <%s> in cache <%s>",
clientId, connectionDescriptors));
}
Channel channel = connectionDescriptors.get(clientId).channel;
//TODO attention channel could be null, because in the mean time it get closed
LOG.trace("Session for clientId {}", clientId);
if (channel.isWritable()) {
LOG.debug("channel is writable");
//if channel is writable don't enqueue
channel.writeAndFlush(pubMessage);
}
}
}