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 
First step on implementing H2 storage

Minor, fixed not imported class

Cleaned MessageStore interface to avoid duplication of methods used in couple. Switched from guid + message to move messages

Separated ISubscriptionsStore from ISessionStore

Implemented H2 queue
This commit is contained in:
andsel
2017-06-02 10:13:34 +02:00
parent c18c915dd2
commit d9ff81131c
39 changed files with 1585 additions and 735 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java
+5 -5
View File
@@ -21,7 +21,7 @@ import io.moquette.spi.ClientSession;
import io.moquette.spi.IMessagesStore;
import io.moquette.spi.ISessionsStore;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.spi.impl.subscriptions.SubscriptionsStore;
import io.moquette.spi.impl.subscriptions.SubscriptionsDirectory;
import io.moquette.spi.impl.subscriptions.Topic;
import io.netty.buffer.ByteBuf;
import io.netty.handler.codec.mqtt.*;
@@ -36,10 +36,10 @@ class MessagesPublisher {
private final ConnectionDescriptorStore connectionDescriptors;
private final ISessionsStore m_sessionsStore;
private final PersistentQueueMessageSender messageSender;
private final SubscriptionsStore subscriptions;
private final SubscriptionsDirectory subscriptions;
public MessagesPublisher(ConnectionDescriptorStore connectionDescriptors, ISessionsStore sessionsStore,
PersistentQueueMessageSender messageSender, SubscriptionsStore subscriptions) {
PersistentQueueMessageSender messageSender, SubscriptionsDirectory subscriptions) {
this.connectionDescriptors = connectionDescriptors;
this.m_sessionsStore = sessionsStore;
this.messageSender = messageSender;
@@ -59,11 +59,11 @@ class MessagesPublisher {
void publish2Subscribers(IMessagesStore.StoredMessage pubMsg, Topic topic, int messageID) {
if (LOG.isTraceEnabled()) {
LOG.trace("Sending publish message to subscribers. CId={}, topic={}, messageId={}, payload={}, " +
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. CId={}, topic={}, messageId={}", pubMsg.getClientID(), topic,
LOG.info("Sending publish message to subscribers. ClientId={}, topic={}, messageId={}", pubMsg.getClientID(), topic,
messageID);
}
publish2Subscribers(pubMsg, topic);
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+29 -30
View File
@@ -24,7 +24,7 @@ import io.moquette.server.netty.NettyUtils;
import io.moquette.spi.*;
import io.moquette.spi.IMessagesStore.StoredMessage;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.spi.impl.subscriptions.SubscriptionsStore;
import io.moquette.spi.impl.subscriptions.SubscriptionsDirectory;
import io.moquette.spi.impl.subscriptions.Topic;
import io.moquette.spi.security.IAuthenticator;
import io.moquette.spi.security.IAuthorizator;
@@ -130,7 +130,8 @@ public class ProtocolProcessor {
protected ConnectionDescriptorStore connectionDescriptors;
protected ConcurrentMap<RunningSubscription, SubscriptionState> subscriptionInCourse;
private SubscriptionsStore subscriptions;
private SubscriptionsDirectory subscriptions;
private ISubscriptionsStore subscriptionStore;
private boolean allowAnonymous;
private boolean allowZeroByteClientId;
private IAuthorizator m_authorizator;
@@ -141,7 +142,6 @@ public class ProtocolProcessor {
private IAuthenticator m_authenticator;
private BrokerInterceptor m_interceptor;
private String m_server_port;
private Qos0PublishHandler qos0PublishHandler;
private Qos1PublishHandler qos1PublishHandler;
@@ -155,23 +155,23 @@ public class ProtocolProcessor {
ProtocolProcessor() {
}
public void init(SubscriptionsStore subscriptions, IMessagesStore storageService, ISessionsStore sessionsStore,
IAuthenticator authenticator, boolean allowAnonymous, IAuthorizator authorizator,
BrokerInterceptor interceptor) {
public void init(SubscriptionsDirectory subscriptions, IMessagesStore storageService, ISessionsStore sessionsStore,
IAuthenticator authenticator, boolean allowAnonymous, IAuthorizator authorizator,
BrokerInterceptor interceptor) {
init(subscriptions, storageService, sessionsStore, authenticator, allowAnonymous, false, authorizator,
interceptor, null);
}
public void init(SubscriptionsStore subscriptions, IMessagesStore storageService, ISessionsStore sessionsStore,
IAuthenticator authenticator, boolean allowAnonymous, boolean allowZeroByteClientId,
IAuthorizator authorizator, BrokerInterceptor interceptor) {
public void init(SubscriptionsDirectory subscriptions, IMessagesStore storageService, ISessionsStore sessionsStore,
IAuthenticator authenticator, boolean allowAnonymous, boolean allowZeroByteClientId,
IAuthorizator authorizator, BrokerInterceptor interceptor) {
init(subscriptions, storageService, sessionsStore, authenticator, allowAnonymous, allowZeroByteClientId,
authorizator, interceptor, null);
}
public void init(SubscriptionsStore subscriptions, IMessagesStore storageService, ISessionsStore sessionsStore,
IAuthenticator authenticator, boolean allowAnonymous, boolean allowZeroByteClientId,
IAuthorizator authorizator, BrokerInterceptor interceptor, String serverPort) {
public void init(SubscriptionsDirectory subscriptions, IMessagesStore storageService, ISessionsStore sessionsStore,
IAuthenticator authenticator, boolean allowAnonymous, boolean allowZeroByteClientId,
IAuthorizator authorizator, BrokerInterceptor interceptor, String serverPort) {
init(new ConnectionDescriptorStore(sessionsStore), subscriptions, storageService, sessionsStore, authenticator,
allowAnonymous, allowZeroByteClientId, authorizator, interceptor, serverPort);
}
@@ -194,7 +194,7 @@ public class ProtocolProcessor {
* @param interceptor
* to notify events to an intercept handler
*/
void init(ConnectionDescriptorStore connectionDescriptors, SubscriptionsStore subscriptions,
void init(ConnectionDescriptorStore connectionDescriptors, SubscriptionsDirectory subscriptions,
IMessagesStore storageService, ISessionsStore sessionsStore, IAuthenticator authenticator,
boolean allowAnonymous, boolean allowZeroByteClientId, IAuthorizator authorizator,
BrokerInterceptor interceptor, String serverPort) {
@@ -212,7 +212,7 @@ public class ProtocolProcessor {
m_authenticator = authenticator;
m_messagesStore = storageService;
m_sessionsStore = sessionsStore;
m_server_port = serverPort;
subscriptionStore = sessionsStore.subscriptionStore();
LOG.info("Initializing messages publisher...");
final PersistentQueueMessageSender messageSender = new PersistentQueueMessageSender(this.connectionDescriptors);
@@ -468,12 +468,11 @@ public class ProtocolProcessor {
LOG.trace("retrieving inflight for messageID <{}>", messageID);
ClientSession targetSession = m_sessionsStore.sessionForClient(clientID);
StoredMessage inflightMsg = targetSession.getInflightMessage(messageID);
targetSession.inFlightAcknowledged(messageID);
StoredMessage inflightMsg = targetSession.inFlightAcknowledged(messageID);
String topic = inflightMsg.getTopic();
m_interceptor
.notifyMessageAcknowledged(new InterceptAcknowledgedMessage(inflightMsg, topic, username, messageID));
InterceptAcknowledgedMessage wrapped = new InterceptAcknowledgedMessage(inflightMsg, topic, username, messageID);
m_interceptor.notifyMessageAcknowledged(wrapped);
}
public static IMessagesStore.StoredMessage asStoredMessage(MqttPublishMessage msg) {
@@ -539,9 +538,9 @@ public class ProtocolProcessor {
} else {
toStoreMsg.setClientID(clientId);
}
if (qos == EXACTLY_ONCE) { // QoS2
guid = m_messagesStore.storePublishForFuture(toStoreMsg);
}
// if (qos == EXACTLY_ONCE) { // QoS2
// guid = m_messagesStore.storePublishForFuture(toStoreMsg);
// }
this.messagesPublisher.publish2Subscribers(toStoreMsg, topic);
if (!msg.fixedHeader().isRetain()) {
@@ -552,11 +551,11 @@ public class ProtocolProcessor {
m_messagesStore.cleanRetained(topic);
return;
}
if (guid == null) {
// before wasn't stored
guid = m_messagesStore.storePublishForFuture(toStoreMsg);
}
m_messagesStore.storeRetained(topic, guid);
// if (guid == null) {
// // before wasn't stored
// guid = m_messagesStore.storePublishForFuture(toStoreMsg);
// }
m_messagesStore.storeRetained(topic, toStoreMsg);
}
/**
@@ -597,7 +596,8 @@ public class ProtocolProcessor {
ClientSession targetSession = m_sessionsStore.sessionForClient(clientID);
// remove from the inflight and move to the QoS2 second phase queue
int messageID = messageId(msg);
targetSession.moveInFlightToSecondPhaseAckWaiting(messageID);
StoredMessage ackedMsg = targetSession.inFlightAcknowledged(messageID);
targetSession.moveInFlightToSecondPhaseAckWaiting(messageID, ackedMsg);
// once received a PUBREC reply with a PUBREL(messageID)
LOG.debug("Processing PUBREC message. CId={}, messageId={}", clientID, messageID);
@@ -678,7 +678,7 @@ public class ProtocolProcessor {
if (descriptor.cleanSession) {
LOG.info("Removing saved subscriptions. CId={}", descriptor.clientID);
m_sessionsStore.wipeSubscriptions(clientID);
subscriptionStore.wipeSubscriptions(clientID);
LOG.info("The saved subscriptions have been removed. CId={}", descriptor.clientID);
}
return true;
@@ -918,9 +918,8 @@ public class ProtocolProcessor {
emptyQueue = true;
} 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.getPayload(),
retained, 0);
msg.isRetained(), 0);
channel.write(pubMsg);
}
}
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java
+11 -16
View File
@@ -25,17 +25,15 @@ import io.moquette.server.config.IResourceLoader;
import io.moquette.spi.IMessagesStore;
import io.moquette.spi.ISessionsStore;
import io.moquette.spi.IStore;
import io.moquette.spi.impl.security.ACLFileParser;
import io.moquette.spi.impl.security.AcceptAllAuthenticator;
import io.moquette.spi.impl.security.DenyAllAuthorizator;
import io.moquette.spi.impl.security.PermitAllAuthorizator;
import io.moquette.spi.impl.security.ResourceAuthenticator;
import io.moquette.spi.ISubscriptionsStore;
import io.moquette.spi.impl.security.*;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.spi.impl.subscriptions.SubscriptionsStore;
import io.moquette.spi.impl.subscriptions.SubscriptionsDirectory;
import io.moquette.spi.security.IAuthenticator;
import io.moquette.spi.security.IAuthorizator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.text.ParseException;
@@ -50,16 +48,13 @@ public class ProtocolProcessorBootstrapper {
private static final Logger LOG = LoggerFactory.getLogger(ProtocolProcessorBootstrapper.class);
public static final String MAPDB_STORE_CLASS = "io.moquette.persistence.MemoryStorageService";
private SubscriptionsStore subscriptions;
private ISessionsStore m_sessionsStore;
private ISubscriptionsStore subscriptionsStore;
private Runnable storeShutdown;
private BrokerInterceptor m_interceptor;
private final ProtocolProcessor m_processor = new ProtocolProcessor();
private ConnectionDescriptorStore connectionDescriptors;
public ProtocolProcessorBootstrapper() {
@@ -86,8 +81,6 @@ public class ProtocolProcessorBootstrapper {
*/
public ProtocolProcessor init(IConfig props, List<? extends InterceptHandler> embeddedObservers,
IAuthenticator authenticator, IAuthorizator authorizator, Server server) {
subscriptions = new SubscriptionsStore();
IMessagesStore messagesStore;
LOG.info("Initializing messages and sessions stores...");
String storageClassName = props.getProperty(BrokerConstants.STORAGE_CLASS_NAME, MAPDB_STORE_CLASS);
@@ -98,6 +91,7 @@ public class ProtocolProcessorBootstrapper {
final IStore store = loadClass(storageClassName, IStore.class, Server.class, server);
messagesStore = store.messagesStore();
m_sessionsStore = store.sessionsStore();
this.subscriptionsStore = m_sessionsStore.subscriptionStore();
storeShutdown = new Runnable() {
@Override
@@ -116,9 +110,10 @@ public class ProtocolProcessorBootstrapper {
observers.add(handler);
}
}
m_interceptor = new BrokerInterceptor(props, observers);
BrokerInterceptor interceptor = new BrokerInterceptor(props, observers);
LOG.info("Initializing subscriptions store...");
SubscriptionsDirectory subscriptions = new SubscriptionsDirectory();
subscriptions.init(m_sessionsStore);
LOG.info("Configuring MQTT authenticator...");
@@ -170,7 +165,7 @@ public class ProtocolProcessorBootstrapper {
boolean allowZeroByteClientId = Boolean
.parseBoolean(props.getProperty(BrokerConstants.ALLOW_ZERO_BYTE_CLIENT_ID_PROPERTY_NAME, "false"));
m_processor.init(connectionDescriptors, subscriptions, messagesStore, m_sessionsStore, authenticator,
allowAnonymous, allowZeroByteClientId, authorizator, m_interceptor,
allowAnonymous, allowZeroByteClientId, authorizator, interceptor,
props.getProperty(BrokerConstants.PORT_PROPERTY_NAME));
return m_processor;
}
@@ -262,7 +257,7 @@ public class ProtocolProcessorBootstrapper {
}
public List<Subscription> getSubscriptions() {
return m_sessionsStore.getSubscriptions();
return this.subscriptionsStore.getSubscriptions();
}
public void shutdown() {
broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java
+2 -9
View File
@@ -19,9 +19,6 @@ package io.moquette.spi.impl;
import io.moquette.server.ConnectionDescriptorStore;
import io.moquette.server.netty.NettyUtils;
import io.moquette.spi.IMessagesStore;
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.moquette.spi.security.IAuthorizator;
import io.netty.channel.Channel;
@@ -31,7 +28,7 @@ import io.netty.handler.codec.mqtt.MqttPubAckMessage;
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;
import static io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader.from;
import static io.netty.handler.codec.mqtt.MqttQoS.AT_MOST_ONCE;
@@ -79,11 +76,7 @@ class Qos1PublishHandler extends QosPublishHandler {
m_messagesStore.cleanRetained(topic);
} else {
// before wasn't stored
MessageGUID guid = m_messagesStore.storePublishForFuture(toStoreMsg);
//TODO Il messaggio non ha il guid settato
// m_messagesStore.storeRetained(topic, toStoreMsg.getGuid());
m_messagesStore.storeRetained(topic, guid);
m_messagesStore.storeRetained(topic, toStoreMsg);
}
}
broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java
+17 -18
View File
@@ -21,16 +21,13 @@ 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.MessageGUID;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.spi.impl.subscriptions.SubscriptionsStore;
import io.moquette.spi.impl.subscriptions.SubscriptionsDirectory;
import io.moquette.spi.impl.subscriptions.Topic;
import io.moquette.spi.security.IAuthorizator;
import io.netty.channel.Channel;
import io.netty.handler.codec.mqtt.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.List;
import static io.moquette.spi.impl.DebugUtils.payload2Str;
import static io.moquette.spi.impl.ProtocolProcessor.asStoredMessage;
@@ -42,14 +39,14 @@ class Qos2PublishHandler extends QosPublishHandler {
private static final Logger LOG = LoggerFactory.getLogger(Qos2PublishHandler.class);
private final SubscriptionsStore subscriptions;
private final SubscriptionsDirectory subscriptions;
private final IMessagesStore m_messagesStore;
private final BrokerInterceptor m_interceptor;
private final ConnectionDescriptorStore connectionDescriptors;
private final ISessionsStore m_sessionsStore;
private final MessagesPublisher publisher;
public Qos2PublishHandler(IAuthorizator authorizator, SubscriptionsStore subscriptions,
public Qos2PublishHandler(IAuthorizator authorizator, SubscriptionsDirectory subscriptions,
IMessagesStore messagesStore, BrokerInterceptor interceptor,
ConnectionDescriptorStore connectionDescriptors, ISessionsStore sessionsStore,
MessagesPublisher messagesPublisher) {
@@ -81,23 +78,22 @@ class Qos2PublishHandler extends QosPublishHandler {
LOG.trace("payload={}, subs Tree={}", payload2Str(toStoreMsg.getPayload()), subscriptions.dumpTree());
}
// QoS2
MessageGUID guid = m_messagesStore.storePublishForFuture(toStoreMsg);
ClientSession sourceSession = m_sessionsStore.sessionForClient(clientID);
sourceSession.markAsInboundInflight(messageID, guid);
m_sessionsStore.sessionForClient(clientID)
.markAsInboundInflight(messageID, toStoreMsg);
sendPubRec(clientID, messageID);
// Next the client will send us a pub rel
// NB publish to subscribers for QoS 2 happen upon PUBREL from publisher
if (msg.fixedHeader().isRetain()) {
if (msg.payload().readableBytes() == 0) {
m_messagesStore.cleanRetained(topic);
} else {
m_messagesStore.storeRetained(topic, guid);
}
}
// if (msg.fixedHeader().isRetain()) {
// if (msg.payload().readableBytes() == 0) {
// m_messagesStore.cleanRetained(topic);
// } else {
// m_messagesStore.storeRetained(topic, toStoreMsg);
// }
// }
//TODO this should happen on PUB_REL, else we notify false positive
m_interceptor.notifyTopicPublished(msg, clientID, username);
}
@@ -123,10 +119,13 @@ class Qos2PublishHandler extends QosPublishHandler {
if (evt.getPayload().readableBytes() == 0) {
m_messagesStore.cleanRetained(topic);
} else {
m_messagesStore.storeRetained(topic, evt.getGuid());
m_messagesStore.storeRetained(topic, evt);
}
}
//TODO here we should notify to the listeners
//m_interceptor.notifyTopicPublished(msg, clientID, username);
sendPubComp(clientID, messageID);
}
broker/src/main/java/io/moquette/spi/impl/subscriptions/Subscription.java
+1 -1
View File
@@ -16,7 +16,7 @@
package io.moquette.spi.impl.subscriptions;
import io.moquette.spi.ISessionsStore.ClientTopicCouple;
import io.moquette.spi.ISubscriptionsStore.ClientTopicCouple;
import io.netty.handler.codec.mqtt.MqttQoS;
import java.io.Serializable;
broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionsStore.java → broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionsDirectory.java
+12 -11
View File
@@ -17,7 +17,8 @@
package io.moquette.spi.impl.subscriptions;
import io.moquette.spi.ISessionsStore;
import io.moquette.spi.ISessionsStore.ClientTopicCouple;
import io.moquette.spi.ISubscriptionsStore;
import io.moquette.spi.ISubscriptionsStore.ClientTopicCouple;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.*;
@@ -27,7 +28,7 @@ import java.util.concurrent.atomic.AtomicReference;
/**
* Represents a tree of topics subscriptions.
*/
public class SubscriptionsStore {
public class SubscriptionsDirectory {
public static class NodeCouple {
@@ -79,8 +80,9 @@ public class SubscriptionsStore {
}
private AtomicReference<TreeNode> subscriptions = new AtomicReference<>(new TreeNode());
private static final Logger LOG = LoggerFactory.getLogger(SubscriptionsStore.class);
private static final Logger LOG = LoggerFactory.getLogger(SubscriptionsDirectory.class);
private volatile ISessionsStore m_sessionsStore;
private volatile ISubscriptionsStore subscriptionsStore;
/**
* Initialize the subscription tree with the list of subscriptions. Maintained for compatibility
@@ -92,21 +94,20 @@ public class SubscriptionsStore {
public void init(ISessionsStore sessionsStore) {
LOG.info("Initializing subscriptions store...");
m_sessionsStore = sessionsStore;
List<ClientTopicCouple> subscriptions = sessionsStore.listAllSubscriptions();
this.subscriptionsStore = sessionsStore.subscriptionStore();
List<ClientTopicCouple> subscriptions = this.subscriptionsStore.listAllSubscriptions();
// reload any subscriptions persisted
if (LOG.isTraceEnabled()) {
LOG.trace("Reloading all stored subscriptions. SubscriptionTree = {}.", dumpTree());
LOG.trace("Reloading all stored subscriptions. SubscriptionTree = {}", dumpTree());
}
for (ClientTopicCouple clientTopic : subscriptions) {
LOG.info(
"Re-subscribing client to topic. ClientId = {}, topicFilter = {}.",
clientTopic.clientID,
clientTopic.topicFilter);
LOG.info("Re-subscribing client to topic CId={}, topicFilter={}", clientTopic.clientID,
clientTopic.topicFilter);
add(clientTopic);
}
if (LOG.isTraceEnabled()) {
LOG.trace("The stored subscriptions have been reloaded. SubscriptionTree = {}.", dumpTree());
LOG.trace("Stored subscriptions have been reloaded. SubscriptionTree = {}", dumpTree());
}
}
@@ -204,7 +205,7 @@ public class SubscriptionsStore {
Map<String, Subscription> subsForClient = new HashMap<>();
for (ClientTopicCouple matchingCouple : matchingSubs) {
Subscription existingSub = subsForClient.get(matchingCouple.clientID);
Subscription sub = m_sessionsStore.getSubscription(matchingCouple);
Subscription sub = this.subscriptionsStore.getSubscription(matchingCouple);
if (sub == null) {
// if the m_sessionStore hasn't the sub because the client disconnected
continue;
broker/src/main/java/io/moquette/spi/impl/subscriptions/TreeNode.java
+1 -1
View File
@@ -16,7 +16,7 @@
package io.moquette.spi.impl.subscriptions;
import io.moquette.spi.ISessionsStore.ClientTopicCouple;
import io.moquette.spi.ISubscriptionsStore.ClientTopicCouple;
import java.util.*;
import java.util.concurrent.LinkedBlockingQueue;