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

Merged topic subscription search and publish into single MessagePublisher method

Browse Source
This commit is contained in:
andsel
2017-04-29 10:59:34 +02:00
parent 67ab24570e
commit 406ffd10db
11 changed files with 94 additions and 119 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java
+6 -6
View File
@@ -44,7 +44,7 @@ class InternalRepublisher {
if (storedMsg.getQos() != MqttQoS.AT_MOST_ONCE) {
LOG.debug("Adding message to inflight zone. ClientId={}, topic={}", targetSession.clientID,
storedMsg.getTopic());
int packetID = targetSession.inFlightAckWaiting(storedMsg.getGuid());
int packetID = targetSession.inFlightAckWaiting(storedMsg);
// set the PacketIdentifier only for QoS > 0
publishMsg = retainedPublish(storedMsg, packetID);
@@ -62,7 +62,7 @@ class InternalRepublisher {
// put in flight zone
LOG.debug("Adding message ot inflight zone. ClientId={}, guid={}, topic={}", clientSession.clientID,
pubEvt.getGuid(), pubEvt.getTopic());
int messageId = clientSession.inFlightAckWaiting(pubEvt.getGuid());
int messageId = clientSession.inFlightAckWaiting(pubEvt);
MqttPublishMessage publishMsg = notRetainedPublish(pubEvt);
// set the PacketIdentifier only for QoS > 0
if (publishMsg.fixedHeader().qosLevel() != MqttQoS.AT_MOST_ONCE) {
@@ -73,22 +73,22 @@ class InternalRepublisher {
}
private MqttPublishMessage notRetainedPublish(IMessagesStore.StoredMessage storedMessage, Integer messageID) {
return createPublishForQos(storedMessage.getTopic(), storedMessage.getQos(), storedMessage.getMessage(), false,
return createPublishForQos(storedMessage.getTopic(), storedMessage.getQos(), storedMessage.getPayload(), false,
messageID);
}
private MqttPublishMessage notRetainedPublish(IMessagesStore.StoredMessage storedMessage) {
return createPublishForQos(storedMessage.getTopic(), storedMessage.getQos(), storedMessage.getMessage(), false,
return createPublishForQos(storedMessage.getTopic(), storedMessage.getQos(), storedMessage.getPayload(), false,
0);
}
private MqttPublishMessage retainedPublish(IMessagesStore.StoredMessage storedMessage) {
return createPublishForQos(storedMessage.getTopic(), storedMessage.getQos(), storedMessage.getMessage(), true,
return createPublishForQos(storedMessage.getTopic(), storedMessage.getQos(), storedMessage.getPayload(), true,
0);
}
private MqttPublishMessage retainedPublish(IMessagesStore.StoredMessage storedMessage, Integer packetID) {
return createPublishForQos(storedMessage.getTopic(), storedMessage.getQos(), storedMessage.getMessage(), true,
return createPublishForQos(storedMessage.getTopic(), storedMessage.getQos(), storedMessage.getPayload(), true,
packetID);
}
broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java
+32 -22
View File
@@ -20,8 +20,9 @@ import io.moquette.server.ConnectionDescriptorStore;
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.moquette.spi.impl.subscriptions.SubscriptionsStore;
import io.moquette.spi.impl.subscriptions.Topic;
import io.netty.buffer.ByteBuf;
import io.netty.handler.codec.mqtt.*;
import org.slf4j.Logger;
@@ -34,15 +35,15 @@ class MessagesPublisher {
private static final Logger LOG = LoggerFactory.getLogger(MessagesPublisher.class);
private final ConnectionDescriptorStore connectionDescriptors;
private final ISessionsStore m_sessionsStore;
private final IMessagesStore m_messagesStore;
private final PersistentQueueMessageSender messageSender;
private final SubscriptionsStore subscriptions;
MessagesPublisher(ConnectionDescriptorStore connectionDescriptors, ISessionsStore sessionsStore,
IMessagesStore messagesStore, PersistentQueueMessageSender messageSender) {
public MessagesPublisher(ConnectionDescriptorStore connectionDescriptors, ISessionsStore sessionsStore,
PersistentQueueMessageSender messageSender, SubscriptionsStore subscriptions) {
this.connectionDescriptors = connectionDescriptors;
this.m_sessionsStore = sessionsStore;
this.m_messagesStore = messagesStore;
this.messageSender = messageSender;
this.subscriptions = subscriptions;
}
static MqttPublishMessage notRetainedPublish(String topic, MqttQoS qos, ByteBuf message) {
@@ -56,36 +57,44 @@ class MessagesPublisher {
return new MqttPublishMessage(fixedHeader, varHeader, message);
}
void publish2Subscribers(IMessagesStore.StoredMessage pubMsg, List<Subscription> topicMatchingSubscriptions) {
final String topic = pubMsg.getTopic();
final MqttQoS publishingQos = pubMsg.getQos();
final ByteBuf origMessage = pubMsg.getMessage();
// if QoS 1 or 2 store the message
// TODO perhaps this block is not needed.
MessageGUID guid = null;
if (publishingQos != MqttQoS.AT_MOST_ONCE) {
guid = m_messagesStore.storePublishForFuture(pubMsg);
void publish2Subscribers(IMessagesStore.StoredMessage pubMsg, Topic topic, int messageID) {
if (LOG.isTraceEnabled()) {
LOG.trace("Sending publish message to subscribers. ClientId={}, topic={}, messageId={}, payload={}, " +
"subscriptionTree={}", pubMsg.getClientID(), topic, messageID, DebugUtils.payload2Str(pubMsg.getPayload()),
subscriptions.dumpTree());
} else {
LOG.info("Sending publish message to subscribers. ClientId={}, topic={}, messageId={}", pubMsg.getClientID(), topic,
messageID);
}
publish2Subscribers(pubMsg, topic);
}
void publish2Subscribers(IMessagesStore.StoredMessage pubMsg, Topic topic) {
List<Subscription> topicMatchingSubscriptions = subscriptions.matches(topic);
final String topic1 = pubMsg.getTopic();
final MqttQoS publishingQos = pubMsg.getQos();
final ByteBuf origPayload = pubMsg.getPayload();
for (final Subscription sub : topicMatchingSubscriptions) {
MqttQoS qos = lowerQosToTheSubscriptionDesired(sub, publishingQos);
ClientSession targetSession = m_sessionsStore.sessionForClient(sub.getClientId());
boolean targetIsActive = this.connectionDescriptors.isConnected(sub.getClientId());
// we need to retain because duplicate only copy r/w indexes and don't retain() causing
// refCnt = 0
ByteBuf message = origMessage.retainedDuplicate();
//TODO move all this logic into messageSender, which puts into the flightZone only the messages that pull out of the queue.
if (targetIsActive) {
LOG.debug("Sending PUBLISH message to active subscriber. CId={}, topicFilter={}, qos={}",
sub.getClientId(), sub.getTopicFilter(), qos);
MqttPublishMessage publishMsg = notRetainedPublish(topic, qos, message);
// we need to retain because duplicate only copy r/w indexes and don't retain() causing
// refCnt = 0
ByteBuf payload = origPayload.retainedDuplicate();
MqttPublishMessage publishMsg;
if (qos != MqttQoS.AT_MOST_ONCE) {
// QoS 1 or 2
int messageId = targetSession.inFlightAckWaiting(guid);
int messageId = targetSession.inFlightAckWaiting(pubMsg);
// set the PacketIdentifier only for QoS > 0
publishMsg = notRetainedPublishWithMessageId(topic, qos, message, messageId);
publishMsg = notRetainedPublishWithMessageId(topic1, qos, payload, messageId);
} else {
publishMsg = notRetainedPublish(topic1, qos, payload);
}
this.messageSender.sendPublish(targetSession, publishMsg);
} else {
@@ -98,4 +107,5 @@ class MessagesPublisher {
}
}
}
}
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+10 -13
View File
@@ -216,16 +216,16 @@ public class ProtocolProcessor {
LOG.info("Initializing messages publisher...");
final PersistentQueueMessageSender messageSender = new PersistentQueueMessageSender(this.connectionDescriptors);
this.messagesPublisher = new MessagesPublisher(connectionDescriptors, sessionsStore, m_messagesStore,
messageSender);
this.messagesPublisher = new MessagesPublisher(connectionDescriptors, sessionsStore, messageSender,
subscriptions);
LOG.info("Initializing QoS publish handlers...");
this.qos0PublishHandler = new Qos0PublishHandler(m_authorizator, subscriptions, m_messagesStore, m_interceptor,
this.qos0PublishHandler = new Qos0PublishHandler(m_authorizator, m_messagesStore, m_interceptor,
this.messagesPublisher);
this.qos1PublishHandler = new Qos1PublishHandler(m_authorizator, subscriptions, m_messagesStore, m_interceptor,
this.connectionDescriptors, m_server_port, this.messagesPublisher);
this.qos1PublishHandler = new Qos1PublishHandler(m_authorizator, m_messagesStore, m_interceptor,
this.connectionDescriptors, this.messagesPublisher);
this.qos2PublishHandler = new Qos2PublishHandler(m_authorizator, subscriptions, m_messagesStore, m_interceptor,
this.connectionDescriptors, m_sessionsStore, m_server_port, this.messagesPublisher);
this.connectionDescriptors, m_sessionsStore, this.messagesPublisher);
LOG.info("Initializing internal republisher...");
this.internalRepublisher = new InternalRepublisher(messageSender);
@@ -542,8 +542,7 @@ public class ProtocolProcessor {
if (qos == EXACTLY_ONCE) { // QoS2
guid = m_messagesStore.storePublishForFuture(toStoreMsg);
}
List<Subscription> topicMatchingSubscriptions = subscriptions.matches(topic);
this.messagesPublisher.publish2Subscribers(toStoreMsg, topicMatchingSubscriptions);
this.messagesPublisher.publish2Subscribers(toStoreMsg, topic);
if (!msg.fixedHeader().isRetain()) {
return;
@@ -564,15 +563,13 @@ public class ProtocolProcessor {
* Specialized version to publish will testament message.
*/
private void forwardPublishWill(WillMessage will, String clientID) {
LOG.info("Publishing will message. CId={}, topic={}", clientID, will.getTopic());
// it has just to publish the message downstream to the subscribers
// NB it's a will publish, it needs a PacketIdentifier for this conn, default to 1
IMessagesStore.StoredMessage tobeStored = asStoredMessage(will);
tobeStored.setClientID(clientID);
Topic topic = new Topic(tobeStored.getTopic());
List<Subscription> topicMatchingSubscriptions = subscriptions.matches(topic);
LOG.info("Publishing will message. CId={}, topic={}", clientID, topic);
this.messagesPublisher.publish2Subscribers(tobeStored, topicMatchingSubscriptions);
this.messagesPublisher.publish2Subscribers(tobeStored, topic);
}
static MqttQoS lowerQosToTheSubscriptionDesired(Subscription sub, MqttQoS qos) {
@@ -922,7 +919,7 @@ public class ProtocolProcessor {
} else {
// recreate a publish from stored publish in queue
boolean retained = m_messagesStore.getMessageByGuid(msg.getGuid()) != null;
MqttPublishMessage pubMsg = createPublishForQos(msg.getTopic(), msg.getQos(), msg.getMessage(),
MqttPublishMessage pubMsg = createPublishForQos( msg.getTopic(), msg.getQos(), msg.getPayload(),
retained, 0);
channel.write(pubMsg);
}
broker/src/main/java/io/moquette/spi/impl/Qos0PublishHandler.java
+7 -23
View File
@@ -18,30 +18,25 @@ package io.moquette.spi.impl;
import io.moquette.server.netty.NettyUtils;
import io.moquette.spi.IMessagesStore;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.spi.impl.subscriptions.SubscriptionsStore;
import io.moquette.spi.impl.subscriptions.Topic;
import io.moquette.spi.security.IAuthorizator;
import io.netty.channel.Channel;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.List;
import static io.moquette.spi.impl.ProtocolProcessor.asStoredMessage;
class Qos0PublishHandler extends QosPublishHandler {
private static final Logger LOG = LoggerFactory.getLogger(Qos0PublishHandler.class);
private final SubscriptionsStore subscriptions;
private final IMessagesStore m_messagesStore;
private final BrokerInterceptor m_interceptor;
private final MessagesPublisher publisher;
Qos0PublishHandler(IAuthorizator authorizator, SubscriptionsStore subscriptions, IMessagesStore messagesStore,
BrokerInterceptor interceptor, MessagesPublisher messagesPublisher) {
public Qos0PublishHandler(IAuthorizator authorizator, IMessagesStore messagesStore, BrokerInterceptor interceptor,
MessagesPublisher messagesPublisher) {
super(authorizator);
this.subscriptions = subscriptions;
this.m_messagesStore = messagesStore;
this.m_interceptor = interceptor;
this.publisher = messagesPublisher;
@@ -50,35 +45,24 @@ class Qos0PublishHandler extends QosPublishHandler {
void receivedPublishQos0(Channel channel, MqttPublishMessage msg) {
// verify if topic can be write
final Topic topic = new Topic(msg.variableHeader().topicName());
if (checkWriteOnTopic(topic, channel)) {
String clientID = NettyUtils.clientID(channel);
String username = NettyUtils.userName(channel);
if (!m_authorizator.canWrite(topic, username, clientID)) {
LOG.error("MQTT client is not authorized to publish on topic. CId={}, topic={}", clientID, topic);
return;
}
// route message to subscribers
IMessagesStore.StoredMessage toStoreMsg = asStoredMessage(msg);
String clientID = NettyUtils.clientID(channel);
toStoreMsg.setClientID(clientID);
final int messageID = msg.variableHeader().messageId();
if (LOG.isTraceEnabled()) {
LOG.trace("Sending publish message to subscribers. ClientId={}, topic={}, messageId={}, payload={}, " +
"subscriptionTree={}", clientID, topic, messageID, DebugUtils.payload2Str(toStoreMsg.getMessage()),
subscriptions.dumpTree());
} else {
LOG.info("Sending publish message to subscribers. ClientId={}, topic={}, messageId={}", clientID, topic,
messageID);
}
List<Subscription> topicMatchingSubscriptions = subscriptions.matches(topic);
this.publisher.publish2Subscribers(toStoreMsg, topicMatchingSubscriptions);
this.publisher.publish2Subscribers(toStoreMsg, topic);
if (msg.fixedHeader().isRetain()) {
// QoS == 0 && retain => clean old retained
m_messagesStore.cleanRetained(topic);
}
String username = NettyUtils.userName(channel);
m_interceptor.notifyTopicPublished(msg, clientID, username);
}
}
broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java
+7 -19
View File
@@ -39,17 +39,14 @@ class Qos1PublishHandler extends QosPublishHandler {
private static final Logger LOG = LoggerFactory.getLogger(Qos1PublishHandler.class);
private final SubscriptionsStore subscriptions;
private final IMessagesStore m_messagesStore;
private final BrokerInterceptor m_interceptor;
private final ConnectionDescriptorStore connectionDescriptors;
private final MessagesPublisher publisher;
Qos1PublishHandler(IAuthorizator authorizator, SubscriptionsStore subscriptions, IMessagesStore messagesStore,
BrokerInterceptor interceptor, ConnectionDescriptorStore connectionDescriptors, String brokerPort,
MessagesPublisher messagesPublisher) {
public Qos1PublishHandler(IAuthorizator authorizator, IMessagesStore messagesStore, BrokerInterceptor interceptor,
ConnectionDescriptorStore connectionDescriptors, MessagesPublisher messagesPublisher) {
super(authorizator);
this.subscriptions = subscriptions;
this.m_messagesStore = messagesStore;
this.m_interceptor = interceptor;
this.connectionDescriptors = connectionDescriptors;
@@ -59,7 +56,10 @@ class Qos1PublishHandler extends QosPublishHandler {
void receivedPublishQos1(Channel channel, MqttPublishMessage msg) {
// verify if topic can be write
final Topic topic = new Topic(msg.variableHeader().topicName());
if (checkWriteOnTopic(topic, channel)) {
String clientID = NettyUtils.clientID(channel);
String username = NettyUtils.userName(channel);
if (!m_authorizator.canWrite(topic, username, clientID)) {
LOG.error("MQTT client is not authorized to publish on topic. CId={}, topic={}", clientID, topic);
return;
}
@@ -67,20 +67,9 @@ class Qos1PublishHandler extends QosPublishHandler {
// route message to subscribers
IMessagesStore.StoredMessage toStoreMsg = asStoredMessage(msg);
String clientID = NettyUtils.clientID(channel);
toStoreMsg.setClientID(clientID);
if (LOG.isTraceEnabled()) {
LOG.trace("Sending publish message to subscribers. ClientId={}, topic={}, messageId={}, payload={}, " +
"subscriptionTree={}", clientID, topic, messageID, DebugUtils.payload2Str(toStoreMsg.getMessage()),
subscriptions.dumpTree());
} else {
LOG.info("Sending publish message to subscribers. ClientId={}, topic={}, messageId={}", clientID, topic,
messageID);
}
List<Subscription> topicMatchingSubscriptions = subscriptions.matches(topic);
this.publisher.publish2Subscribers(toStoreMsg, topicMatchingSubscriptions);
this.publisher.publish2Subscribers(toStoreMsg, topic, messageID);
sendPubAck(clientID, messageID);
@@ -94,7 +83,6 @@ class Qos1PublishHandler extends QosPublishHandler {
}
}
String username = NettyUtils.userName(channel);
m_interceptor.notifyTopicPublished(msg, clientID, username);
}
broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java
+11 -19
View File
@@ -49,10 +49,10 @@ class Qos2PublishHandler extends QosPublishHandler {
private final ISessionsStore m_sessionsStore;
private final MessagesPublisher publisher;
Qos2PublishHandler(IAuthorizator authorizator, SubscriptionsStore subscriptions,
IMessagesStore messagesStore, BrokerInterceptor interceptor,
ConnectionDescriptorStore connectionDescriptors, ISessionsStore sessionsStore, String brokerPort,
MessagesPublisher messagesPublisher) {
public Qos2PublishHandler(IAuthorizator authorizator, SubscriptionsStore subscriptions,
IMessagesStore messagesStore, BrokerInterceptor interceptor,
ConnectionDescriptorStore connectionDescriptors, ISessionsStore sessionsStore,
MessagesPublisher messagesPublisher) {
super(authorizator);
this.subscriptions = subscriptions;
this.m_messagesStore = messagesStore;
@@ -65,18 +65,20 @@ class Qos2PublishHandler extends QosPublishHandler {
void receivedPublishQos2(Channel channel, MqttPublishMessage msg) {
final Topic topic = new Topic(msg.variableHeader().topicName());
// check if the topic can be wrote
if (checkWriteOnTopic(topic, channel)) {
String clientID = NettyUtils.clientID(channel);
String username = NettyUtils.userName(channel);
if (!m_authorizator.canWrite(topic, username, clientID)) {
LOG.error("MQTT client is not authorized to publish on topic. CId={}, topic={}", clientID, topic);
return;
}
final int messageID = msg.variableHeader().messageId();
IMessagesStore.StoredMessage toStoreMsg = asStoredMessage(msg);
String clientID = NettyUtils.clientID(channel);
toStoreMsg.setClientID(clientID);
LOG.info("Sending publish message to subscribers CId={}, topic={}, messageId={}", clientID, topic, messageID);
if (LOG.isTraceEnabled()) {
LOG.trace("payload={}, subs Tree={}", payload2Str(toStoreMsg.getMessage()), subscriptions.dumpTree());
LOG.trace("payload={}, subs Tree={}", payload2Str(toStoreMsg.getPayload()), subscriptions.dumpTree());
}
// QoS2
@@ -96,7 +98,6 @@ class Qos2PublishHandler extends QosPublishHandler {
m_messagesStore.storeRetained(topic, guid);
}
}
String username = NettyUtils.userName(channel);
m_interceptor.notifyTopicPublished(msg, clientID, username);
}
@@ -115,20 +116,11 @@ class Qos2PublishHandler extends QosPublishHandler {
throw new IllegalArgumentException("Can't find inbound inflight message");
}
final Topic topic = new Topic(evt.getTopic());
List<Subscription> topicMatchingSubscriptions = subscriptions.matches(topic);
if (LOG.isDebugEnabled()) {
LOG.debug("Sending publish message to subscribers. CId={}, topic={}, messageId={}, payload={}, " +
"subscriptionTree={}", clientID, topic, messageID, payload2Str(evt.getMessage()),
subscriptions.dumpTree());
} else {
LOG.info("Sending publish message to subscribers. CId={}, topic={}, messageId={}", clientID, topic,
messageID);
}
this.publisher.publish2Subscribers(evt, topicMatchingSubscriptions);
this.publisher.publish2Subscribers(evt, topic, messageID);
if (evt.isRetained()) {
if (evt.getMessage().readableBytes() == 0) {
if (evt.getPayload().readableBytes() == 0) {
m_messagesStore.cleanRetained(topic);
} else {
m_messagesStore.storeRetained(topic, evt.getGuid());
broker/src/main/java/io/moquette/spi/impl/QosPublishHandler.java
+2 -5
View File
@@ -27,7 +27,7 @@ abstract class QosPublishHandler {
private static final Logger LOG = LoggerFactory.getLogger(QosPublishHandler.class);
private final IAuthorizator m_authorizator;
protected final IAuthorizator m_authorizator;
protected QosPublishHandler(IAuthorizator m_authorizator) {
this.m_authorizator = m_authorizator;
@@ -37,10 +37,7 @@ abstract class QosPublishHandler {
String clientID = NettyUtils.clientID(channel);
String username = NettyUtils.userName(channel);
if (!m_authorizator.canWrite(topic, username, clientID)) {
LOG.error(
"The MQTT client is not authorized to publish on topic. MqttClientId = {}, topic = {}.",
clientID,
topic);
LOG.error("MQTT client is not authorized to publish on topic. CId={}, topic={}", clientID, topic);
return true;
}
return false;
broker/src/main/java/io/moquette/spi/impl/subscriptions/Topic.java
+7
View File
@@ -167,4 +167,11 @@ public class Topic implements Serializable {
public int hashCode() {
return topic.hashCode();
}
/**
* Factory method
* */
public static Topic asTopic(String s) {
return new Topic(s);
}
}