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

Big refactory on session management:

Browse Source 
- introduced a unique point of access and management of sessions state (SessionRepository)
- separated behaiour for sessions with clean flag to false into DurableSession
- separated behaviour for session with clean flag to true into TransientSession
This commit is contained in:
andsel
2017-10-14 17:37:26 +02:00
parent 6bacce6cec
commit 1b272e5849
43 changed files with 1107 additions and 1710 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/connections/IConnectionsManager.java
-8
View File
@@ -55,14 +55,6 @@ public interface IConnectionsManager {
*/
boolean closeConnection(String clientID, boolean closeImmediately);
/**
* Returns the state of the session of a given client.
*
* @param clientID
* @return
*/
MqttSession getSessionStatus(String clientID);
/**
* Returns the state of all the sessions
*
broker/src/main/java/io/moquette/persistence/MemorySessionStore.java
+49 -96
View File
@@ -17,7 +17,6 @@
package io.moquette.persistence;
import io.moquette.server.Constants;
import io.moquette.spi.ClientSession;
import io.moquette.spi.IMessagesStore.StoredMessage;
import io.moquette.spi.ISessionsStore;
import io.moquette.spi.ISubscriptionsStore;
@@ -30,7 +29,6 @@ import java.util.*;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicReference;
public class MemorySessionStore implements ISessionsStore, ISubscriptionsStore {
@@ -38,24 +36,23 @@ public class MemorySessionStore implements ISessionsStore, ISubscriptionsStore {
class Session {
final String clientID;
final ClientSession clientSession;
final Map<Topic, Subscription> subscriptions = new ConcurrentHashMap<>();
final AtomicReference<PersistentSession> persistentSession = new AtomicReference<>(null);
boolean cleanSession;
final BlockingQueue<StoredMessage> queue = new ArrayBlockingQueue<>(Constants.MAX_MESSAGE_QUEUE);
final Map<Integer, StoredMessage> secondPhaseStore = new ConcurrentHashMap<>();
final Map<Integer, StoredMessage> outboundFlightMessages =
Collections.synchronizedMap(new HashMap<Integer, StoredMessage>());
final Map<Integer, StoredMessage> inboundFlightMessages = new ConcurrentHashMap<>();
Session(String clientID, ClientSession clientSession) {
Session(String clientID, boolean cleanSession) {
this.clientID = clientID;
this.clientSession = clientSession;
this.cleanSession = cleanSession;
}
}
private final Map<String, Session> sessions = new ConcurrentHashMap<>();
public MemorySessionStore() {
MemorySessionStore() {
}
private Session getSession(String clientID) {
@@ -110,64 +107,57 @@ public class MemorySessionStore implements ISessionsStore, ISubscriptionsStore {
}
@Override
public ClientSession createNewSession(String clientID, boolean cleanSession) {
LOG.debug("createNewSession for client <{}>", clientID);
Session session = sessions.get(clientID);
if (session != null) {
LOG.error("already exists a session for client <{}>, bad condition", clientID);
throw new IllegalArgumentException("Can't create a session with the ID of an already existing" + clientID);
}
LOG.debug("clientID {} is a newcome, creating it's empty subscriptions set", clientID);
session = new Session(clientID, new ClientSession(clientID, this, this, cleanSession));
session.persistentSession.set(new PersistentSession(cleanSession));
sessions.put(clientID, session);
return session.clientSession;
public void createNewDurableSession(String clientID) {
Session innerSession = new Session(clientID, false);
sessions.put(clientID, innerSession);
}
@Override
public ClientSession sessionForClient(String clientID) {
if (!sessions.containsKey(clientID)) {
LOG.error("Can't find the session for client <{}>", clientID);
return null;
}
PersistentSession storedSession = sessions.get(clientID).persistentSession.get();
return new ClientSession(clientID, this, this, storedSession.cleanSession);
public void removeDurableSession(String clientId) {
this.sessions.remove(clientId);
this.wipeSubscriptions(clientId);
}
@Override
public Collection<ClientSession> getAllSessions() {
Collection<ClientSession> result = new ArrayList<>();
public void updateCleanStatus(String clientId, boolean newCleanStatus) {
sessions.get(clientId).cleanSession = newCleanStatus;
}
@Override
public PersistentSession loadSessionByKey(String clientID) {
return new PersistentSession(clientID, sessions.get(clientID).cleanSession);
}
@Override
public Collection<PersistentSession> listAllSessions() {
Collection<PersistentSession> result = new ArrayList<>();
for (Session entry : sessions.values()) {
result.add(new ClientSession(entry.clientID, this, this, entry.persistentSession.get().cleanSession));
result.add(new PersistentSession(entry.clientID, entry.cleanSession));
}
return result;
}
@Override
public void updateCleanStatus(String clientID, boolean cleanSession) {
if (!sessions.containsKey(clientID)) {
LOG.error("Can't find the session for client <{}>", clientID);
return;
}
sessions.get(clientID).persistentSession.set(new PersistentSession(cleanSession));
}
@Override
public List<ClientTopicCouple> listAllSubscriptions() {
List<ClientTopicCouple> allSubscriptions = new ArrayList<>();
public List<Subscription> listAllSubscriptions() {
List<Subscription> allSubscriptions = new ArrayList<>();
for (Session entry : sessions.values()) {
for (Subscription sub : entry.subscriptions.values()) {
allSubscriptions.add(sub.asClientTopicCouple());
}
allSubscriptions.addAll(entry.subscriptions.values());
}
return allSubscriptions;
}
@Override
public Subscription getSubscription(ClientTopicCouple couple) {
String clientID = couple.clientID;
public Collection<Subscription> listClientSubscriptions(String clientID) {
final Session session = sessions.get(clientID);
if (session == null) {
throw new IllegalStateException("Asking for subscriptions of not persisted client: " + clientID);
}
return session.subscriptions.values();
}
@Override
public Subscription reload(Subscription subcription) {
String clientID = subcription.getClientId();
if (!sessions.containsKey(clientID)) {
LOG.error("Can't find the session for client <{}>", clientID);
return null;
@@ -177,17 +167,9 @@ public class MemorySessionStore implements ISessionsStore, ISubscriptionsStore {
if (subscriptions == null || subscriptions.isEmpty()) {
return null;
}
return subscriptions.get(couple.topicFilter);
return subscriptions.get(subcription.getTopicFilter());
}
@Override
public List<Subscription> getSubscriptions() {
List<Subscription> subscriptions = new ArrayList<>();
for (Session entry : sessions.values()) {
subscriptions.addAll(entry.subscriptions.values());
}
return subscriptions;
}
@Override
public StoredMessage inFlightAck(String clientID, int messageID) {
@@ -234,7 +216,10 @@ public class MemorySessionStore implements ISessionsStore, ISubscriptionsStore {
@Override
public void dropQueue(String clientID) {
sessions.get(clientID).queue.clear();
final Session session = sessions.get(clientID);
session.queue.clear();
session.outboundFlightMessages.clear();
session.inboundFlightMessages.clear();
}
@Override
@@ -251,7 +236,7 @@ public class MemorySessionStore implements ISessionsStore, ISubscriptionsStore {
}
@Override
public StoredMessage secondPhaseAcknowledged(String clientID, int messageID) {
public StoredMessage completeReleasedPublish(String clientID, int messageID) {
LOG.info("Acknowledged message in second phase, clientID <{}> messageID {}", clientID, messageID);
return getSession(clientID).secondPhaseStore.remove(messageID);
}
@@ -264,35 +249,13 @@ public class MemorySessionStore implements ISessionsStore, ISubscriptionsStore {
return 0;
}
return session.inboundFlightMessages.size() + session.secondPhaseStore.size()
+ session.outboundFlightMessages.size();
return session.inboundFlightMessages.size() +
// countPubReleaseWaitingPubComplete(clientID) +
session.outboundFlightMessages.size();
}
@Override
public StoredMessage inboundInflight(String clientID, int messageID) {
return getSession(clientID).inboundFlightMessages.get(messageID);
}
@Override
public void markAsInboundInflight(String clientID, int messageID, StoredMessage msg) {
if (!sessions.containsKey(clientID))
LOG.error("Can't find the session for client <{}>", clientID);
sessions.get(clientID).inboundFlightMessages.put(messageID, msg);
}
@Override
public int getPendingPublishMessagesNo(String clientID) {
if (!sessions.containsKey(clientID)) {
LOG.error("Can't find the session for client <{}>", clientID);
return 0;
}
return sessions.get(clientID).queue.size();
}
@Override
public int getSecondPhaseAckPendingMessages(String clientID) {
public int countPubReleaseWaitingPubComplete(String clientID) {
if (!sessions.containsKey(clientID)) {
LOG.error("Can't find the session for client <{}>", clientID);
return 0;
@@ -302,7 +265,7 @@ public class MemorySessionStore implements ISessionsStore, ISubscriptionsStore {
}
@Override
public void cleanSession(String clientID) {
public void removeTemporaryQoS2(String clientID) {
LOG.debug("Session cleanup for client <{}>", clientID);
Session session = sessions.get(clientID);
@@ -311,18 +274,8 @@ public class MemorySessionStore implements ISessionsStore, ISubscriptionsStore {
return;
}
// remove also the messages stored of type QoS1/2
LOG.info("Removing stored messages with QoS 1 and 2. ClientId={}", clientID);
LOG.info("Removing stored messages with QoS 2. ClientId={}", clientID);
session.secondPhaseStore.clear();
session.outboundFlightMessages.clear();
session.inboundFlightMessages.clear();
LOG.info("Wiping existing subscriptions. ClientId={}", clientID);
wipeSubscriptions(clientID);
//remove also the enqueued messages
dropQueue(clientID);
// TODO this missing last step breaks the junit test
//sessions.remove(clientID);
broker/src/main/java/io/moquette/persistence/PersistentSession.java
+3 -1
View File
@@ -25,8 +25,10 @@ public class PersistentSession implements Serializable {
private static final long serialVersionUID = 5052054783220481854L;
public final boolean cleanSession;
public final String clientID;
public PersistentSession(boolean cleanSession) {
public PersistentSession(String clientID, boolean cleanSession) {
this.clientID = clientID;
this.cleanSession = cleanSession;
}
}
broker/src/main/java/io/moquette/server/ConnectionDescriptor.java
+1 -1
View File
@@ -126,7 +126,7 @@ public class ConnectionDescriptor {
@Override
public String toString() {
return "ConnectionDescriptor{" + "clientID=" + clientID + ", cleanSession=" + cleanSession + ", state="
return "ConnectionDescriptor{" + "clientID=" + clientID + ", removeTemporaryQoS2=" + cleanSession + ", state="
+ channelState.get() + '}';
}
broker/src/main/java/io/moquette/server/ConnectionDescriptorStore.java
+7 -17
View File
@@ -23,12 +23,13 @@ import io.moquette.connections.MqttSubscription;
import io.moquette.server.netty.metrics.BytesMetrics;
import io.moquette.server.netty.metrics.MessageMetrics;
import io.moquette.spi.ClientSession;
import io.moquette.spi.ISessionsStore;
import io.moquette.spi.impl.SessionsRepository;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttMessageType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.Collection;
import java.util.concurrent.ConcurrentHashMap;
@@ -39,11 +40,11 @@ public class ConnectionDescriptorStore implements IConnectionsManager {
private static final Logger LOG = LoggerFactory.getLogger(ConnectionDescriptorStore.class);
private final ConcurrentMap<String, ConnectionDescriptor> connectionDescriptors;
private final ISessionsStore sessionsStore;
private final SessionsRepository sessionsRepository;
public ConnectionDescriptorStore(ISessionsStore sessionsStore) {
public ConnectionDescriptorStore(SessionsRepository sessionsRepository) {
this.connectionDescriptors = new ConcurrentHashMap<>();
this.sessionsStore = sessionsStore;
this.sessionsRepository = sessionsRepository;
}
public boolean sendMessage(MqttMessage message, Integer messageID, String clientID) {
@@ -124,22 +125,11 @@ public class ConnectionDescriptorStore implements IConnectionsManager {
}
}
@Override
public MqttSession getSessionStatus(String clientID) {
LOG.info("Retrieving status of session. CId=<{}>", clientID);
ClientSession session = sessionsStore.sessionForClient(clientID);
if (session == null) {
LOG.error("MQTT client ID doesn't have an associated session. CId=<{}>", clientID);
return null;
}
return buildMqttSession(session);
}
@Override
public Collection<MqttSession> getSessions() {
LOG.info("Retrieving status of all sessions.");
Collection<MqttSession> result = new ArrayList<>();
for (ClientSession session : sessionsStore.getAllSessions()) {
for (ClientSession session : this.sessionsRepository.getAllSessions()) {
result.add(buildMqttSession(session));
}
return result;
@@ -165,7 +155,7 @@ public class ConnectionDescriptorStore implements IConnectionsManager {
result.setConnectionEstablished(false);
}
result.setPendingPublishMessagesNo(session.getPendingPublishMessagesNo());
result.setSecondPhaseAckPendingMessages(session.getSecondPhaseAckPendingMessages());
result.setSecondPhaseAckPendingMessages(session.countPubReleaseWaitingPubComplete());
result.setInflightMessages(session.getInflightMessagesNo());
return result;
}
broker/src/main/java/io/moquette/server/Constants.java
+1 -1
View File
@@ -22,7 +22,7 @@ package io.moquette.server;
public final class Constants {
public static final String ATTR_CLIENTID = "ClientID";
public static final String CLEAN_SESSION = "cleanSession";
public static final String CLEAN_SESSION = "removeTemporaryQoS2";
public static final String KEEP_ALIVE = "keepAlive";
public static final int MAX_MESSAGE_QUEUE = 1024; // number of messages
broker/src/main/java/io/moquette/spi/ClientSession.java
+40 -163
View File
@@ -16,14 +16,10 @@
package io.moquette.spi;
import io.moquette.spi.ISubscriptionsStore.ClientTopicCouple;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.spi.impl.subscriptions.Topic;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.HashSet;
import java.util.Queue;
import java.util.Set;
/**
@@ -40,164 +36,51 @@ import java.util.Set;
* <li>Optionally, QoS 0 messages pending transmission to the Client.</li>
* </ul>
*/
public class ClientSession {
class OutboundFlightZone {
/**
* Save the binding messageID, clientID - message
*
* @param messageID the packet ID used in transmission
* @param msg the message to put in flight zone
*/
void waitingAck(int messageID, IMessagesStore.StoredMessage msg) {
if (LOG.isTraceEnabled()) {
LOG.trace("Adding to inflight {}, guid <{}>", messageID, msg.getGuid());
}
m_sessionsStore.inFlight(ClientSession.this.clientID, messageID, msg);
}
IMessagesStore.StoredMessage acknowledged(int messageID) {
if (LOG.isTraceEnabled())
LOG.trace("Acknowledging inflight, clientID <{}> messageID {}", ClientSession.this.clientID, messageID);
return m_sessionsStore.inFlightAck(ClientSession.this.clientID, messageID);
}
}
class InboundFlightZone {
public IMessagesStore.StoredMessage lookup(int messageID) {
return m_sessionsStore.inboundInflight(clientID, messageID);
}
public void waitingRel(int messageID, IMessagesStore.StoredMessage msg) {
m_sessionsStore.markAsInboundInflight(clientID, messageID, msg);
}
}
private static final Logger LOG = LoggerFactory.getLogger(ClientSession.class);
public abstract class ClientSession {
public final String clientID;
private final ISessionsStore m_sessionsStore;
protected Set<Subscription> subscriptions = new HashSet<>();
private Set<Subscription> subscriptions = new HashSet<>();
private final ISubscriptionsStore subscriptionsStore;
private volatile boolean cleanSession;
private final OutboundFlightZone outboundFlightZone;
private final InboundFlightZone inboundFlightZone;
public ClientSession(String clientID, ISessionsStore sessionsStore, ISubscriptionsStore subscriptionsStore,
boolean cleanSession) {
public ClientSession(String clientID) {
this.clientID = clientID;
this.m_sessionsStore = sessionsStore;
this.subscriptionsStore = subscriptionsStore;
this.cleanSession = cleanSession;
this.outboundFlightZone = new OutboundFlightZone();
this.inboundFlightZone = new InboundFlightZone();
}
/**
* Return the list of persisted publishes for the given clientID. For QoS1 and QoS2 with clean
* session flag, this method return the list of missed publish events while the client was
* disconnected.
*
* @return the list of messages to be delivered for client related to the session.
*/
public Queue<IMessagesStore.StoredMessage> queue() {
LOG.info("Retrieving enqueued messages. CId={}", clientID);
return this.m_sessionsStore.queue(clientID);
}
@Override
public String toString() {
return "ClientSession{clientID='" + clientID + "'}";
}
public boolean subscribe(Subscription newSubscription) {
LOG.info("Adding new subscription. CId={}, topics={}, qos={}", newSubscription.getClientId(),
newSubscription.getTopicFilter(), newSubscription.getRequestedQos());
boolean validTopic = newSubscription.getTopicFilter().isValid();
if (!validTopic) {
LOG.warn("The topic filter is not valid. CId={}, topics={}", newSubscription.getClientId(),
newSubscription.getTopicFilter());
// send SUBACK with 0x80 for this topic filter
return false;
}
ClientTopicCouple matchingCouple = new ClientTopicCouple(this.clientID, newSubscription.getTopicFilter());
Subscription existingSub = subscriptionsStore.getSubscription(matchingCouple);
// update the selected subscriptions if not present or if has a greater qos
if (existingSub == null || existingSub.getRequestedQos().value() < newSubscription.getRequestedQos().value()) {
if (existingSub != null) {
LOG.info("Subscription already existed with a lower QoS value. It will be updated. CId={}, " +
"topics={}, existingQos={}, newQos={}", newSubscription.getClientId(),
newSubscription.getTopicFilter(), existingSub.getRequestedQos(), newSubscription.getRequestedQos());
subscriptions.remove(newSubscription);
}
subscriptions.add(newSubscription);
subscriptionsStore.addNewSubscription(newSubscription);
}
return true;
}
public void unsubscribeFrom(Topic topicFilter) {
LOG.info("Removing subscription. CId={}, topics={}", clientID, topicFilter);
subscriptionsStore.removeSubscription(topicFilter, clientID);
Set<Subscription> subscriptionsToRemove = new HashSet<>();
public Subscription findSubscriptionByTopicFilter(Subscription matchingSub) {
for (Subscription sub : this.subscriptions) {
if (sub.getTopicFilter().equals(topicFilter)) {
subscriptionsToRemove.add(sub);
if (matchingSub.equals(sub)) {
return sub;
}
}
subscriptions.removeAll(subscriptionsToRemove);
return null;
}
public void disconnect() {
if (this.cleanSession) {
LOG.info("Client disconnected. Removing its subscriptions. CId={}", clientID);
// cleanup topic subscriptions
cleanSession();
}
public Set<Subscription> getSubscriptions() {
return subscriptions;
}
public void cleanSession() {
m_sessionsStore.cleanSession(this.clientID);
}
public abstract boolean isCleanSession();
public boolean isCleanSession() {
return this.cleanSession;
}
public abstract boolean subscribe(Subscription newSubscription);
public void cleanSession(boolean cleanSession) {
this.cleanSession = cleanSession;
this.m_sessionsStore.updateCleanStatus(this.clientID, cleanSession);
}
public abstract void unsubscribeFrom(Topic topicFilter);
public int nextPacketId() {
return this.m_sessionsStore.nextPacketID(this.clientID);
}
public abstract void disconnect();
public IMessagesStore.StoredMessage inFlightAcknowledged(int messageID) {
return outboundFlightZone.acknowledged(messageID);
}
public abstract void cleanSession();
protected abstract int nextPacketId();
public abstract IMessagesStore.StoredMessage inFlightAcknowledged(int messageID);
/**
* Mark the message identified by guid as publish in flight.
* Mark the message as publish in flight.
*
* @return the packetID for the message in flight.
* */
public int inFlightAckWaiting(IMessagesStore.StoredMessage msg) {
LOG.debug("Adding message ot inflight zone. CId={}", clientID);
int messageId = ClientSession.this.nextPacketId();
outboundFlightZone.waitingAck(messageId, msg);
return messageId;
}
public abstract int inFlightAckWaiting(IMessagesStore.StoredMessage msg);
public IMessagesStore.StoredMessage secondPhaseAcknowledged(int messageID) {
return m_sessionsStore.secondPhaseAcknowledged(clientID, messageID);
}
public abstract IMessagesStore.StoredMessage completeReleasedPublish(int messageID);
/**
* Enqueue a message to be sent to the client.
@@ -205,37 +88,31 @@ public class ClientSession {
* @param message
* the message to enqueue.
*/
public void enqueue(IMessagesStore.StoredMessage message) {
this.m_sessionsStore.queue(this.clientID).add(message);
}
public abstract void enqueue(IMessagesStore.StoredMessage message);
public IMessagesStore.StoredMessage inboundInflight(int messageID) {
return inboundFlightZone.lookup(messageID);
}
/**
* @return the next message queued to this client or null if empty.
* */
public abstract EnqueuedMessage poll();
public void markAsInboundInflight(int messageID, IMessagesStore.StoredMessage msg) {
inboundFlightZone.waitingRel(messageID, msg);
}
public abstract boolean isEmptyQueue();
public void moveInFlightToSecondPhaseAckWaiting(int messageID, IMessagesStore.StoredMessage msg) {
m_sessionsStore.moveInFlightToSecondPhaseAckWaiting(this.clientID, messageID, msg);
}
public abstract void dropQueue();
public Set<Subscription> getSubscriptions() {
//TODO WARN this doesn'load anything from DB
return subscriptions;
}
public abstract IMessagesStore.StoredMessage inboundInflight(int messageID);
public int getPendingPublishMessagesNo() {
return m_sessionsStore.getPendingPublishMessagesNo(clientID);
}
public abstract void markAsInboundInflight(int messageID, IMessagesStore.StoredMessage msg);
public int getSecondPhaseAckPendingMessages() {
return m_sessionsStore.getSecondPhaseAckPendingMessages(clientID);
}
public abstract void moveInFlightToSecondPhaseAckWaiting(int messageID, IMessagesStore.StoredMessage msg);
public int getInflightMessagesNo() {
return m_sessionsStore.getInflightMessagesNo(clientID);
}
public abstract int getPendingPublishMessagesNo();
public abstract int countPubReleaseWaitingPubComplete();
public abstract int getInflightMessagesNo();
@Override
public String toString() {
return getClass().getSimpleName() + "{clientID='" + clientID + "'}";
}
}
broker/src/main/java/io/moquette/spi/DurableSession.java
+229
View File
@@ -0,0 +1,229 @@
/*
* Copyright (c) 2012-2017 The original author or authors
* ------------------------------------------------------
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
* and Apache License v2.0 which accompanies this distribution.
*
* The Eclipse Public License is available at
* http://www.eclipse.org/legal/epl-v10.html
*
* The Apache License v2.0 is available at
* http://www.opensource.org/licenses/apache2.0.php
*
* You may elect to redistribute this code under either of these licenses.
*/
package io.moquette.spi;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.spi.impl.subscriptions.Topic;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Collection;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
public class DurableSession extends ClientSession {
class OutboundFlightZone {
/**
* Save the binding messageID, clientID - message
*
* @param messageID the packet ID used in transmission
* @param msg the message to put in flight zone
*/
void waitingAck(int messageID, IMessagesStore.StoredMessage msg) {
if (LOG.isTraceEnabled()) {
LOG.trace("Adding to inflight {}, guid <{}>", messageID, msg.getGuid());
}
sessionsStore.inFlight(clientID, messageID, msg);
}
IMessagesStore.StoredMessage acknowledged(int messageID) {
if (LOG.isTraceEnabled()) {
LOG.trace("Acknowledging inflight, clientID <{}> messageID {}", clientID, messageID);
}
return sessionsStore.inFlightAck(clientID, messageID);
}
}
class InboundFlightZone {
final Map<Integer, IMessagesStore.StoredMessage> inboundFlightMessages = new ConcurrentHashMap<>();
IMessagesStore.StoredMessage lookup(int messageID) {
return inboundFlightMessages.get(messageID);
}
void waitingRel(int messageID, IMessagesStore.StoredMessage msg) {
inboundFlightMessages.put(messageID, msg);
}
}
private static final Logger LOG = LoggerFactory.getLogger(DurableSession.class);
private final ISessionsStore sessionsStore;
private final ISubscriptionsStore subscriptionsStore;
private final OutboundFlightZone outboundFlightZone;
private final InboundFlightZone inboundFlightZone;
public DurableSession(String clientID, ISessionsStore sessions, ISubscriptionsStore subscriptionsStore) {
super(clientID);
this.subscriptionsStore = subscriptionsStore;
this.sessionsStore = sessions;
this.outboundFlightZone = new OutboundFlightZone();
this.inboundFlightZone = new InboundFlightZone();
}
public void reloadAllSubscriptionsFromStore() {
Collection<Subscription> reloadedSubscriptions = this.subscriptionsStore.listClientSubscriptions(this.clientID);
this.subscriptions.addAll(reloadedSubscriptions);
}
@Override
public boolean isCleanSession() {
return false;
}
@Override
public boolean subscribe(Subscription newSubscription) {
LOG.info("Adding new subscription. CId={}, topics={}, qos={}", newSubscription.getClientId(),
newSubscription.getTopicFilter(), newSubscription.getRequestedQos());
boolean validTopic = newSubscription.getTopicFilter().isValid();
if (!validTopic) {
LOG.warn("The topic filter is not valid. CId={}, topics={}", newSubscription.getClientId(),
newSubscription.getTopicFilter());
// send SUBACK with 0x80 for this topic filter
return false;
}
Subscription existingSub = subscriptionsStore.reload(newSubscription);
// update the selected subscriptions if not present or if has a greater qos
if (existingSub == null || existingSub.qosLessThan(newSubscription)) {
if (existingSub != null) {
LOG.info("Subscription already existed with a lower QoS value. It will be updated. CId={}, " +
"topics={}, existingQos={}, newQos={}", newSubscription.getClientId(),
newSubscription.getTopicFilter(), existingSub.getRequestedQos(), newSubscription.getRequestedQos());
subscriptions.remove(newSubscription);
}
subscriptions.add(newSubscription);
subscriptionsStore.addNewSubscription(newSubscription);
}
return true;
}
@Override
public void unsubscribeFrom(Topic topicFilter) {
LOG.info("Removing subscription. CId={}, topics={}", clientID, topicFilter);
subscriptionsStore.removeSubscription(topicFilter, clientID);
Set<Subscription> subscriptionsToRemove = new HashSet<>();
for (Subscription sub : this.subscriptions) {
if (sub.getTopicFilter().equals(topicFilter)) {
subscriptionsToRemove.add(sub);
}
}
subscriptions.removeAll(subscriptionsToRemove);
}
@Override
public void cleanSession() {
sessionsStore.removeTemporaryQoS2(this.clientID);
LOG.info("Wiping existing subscriptions. ClientId={}", clientID);
subscriptionsStore.wipeSubscriptions(clientID);
LOG.info("Removing queues. ClientId={}", clientID);
sessionsStore.dropQueue(clientID);
}
@Override
public void disconnect() {
LOG.info("Client disconnected. Removing its subscriptions. CId={}", clientID);
// cleanup topic subscriptions
cleanSession();
}
@Override
protected int nextPacketId() {
return this.sessionsStore.nextPacketID(this.clientID);
}
@Override
public IMessagesStore.StoredMessage inFlightAcknowledged(int messageID) {
return outboundFlightZone.acknowledged(messageID);
}
@Override
public int inFlightAckWaiting(IMessagesStore.StoredMessage msg) {
LOG.debug("Adding message to inflight zone. CId={}", clientID);
int messageId = this.nextPacketId();
outboundFlightZone.waitingAck(messageId, msg);
return messageId;
}
@Override
public IMessagesStore.StoredMessage inboundInflight(int messageID) {
return inboundFlightZone.lookup(messageID);
}
@Override
public void markAsInboundInflight(int messageID, IMessagesStore.StoredMessage msg) {
inboundFlightZone.waitingRel(messageID, msg);
}
@Override
public void moveInFlightToSecondPhaseAckWaiting(int messageID, IMessagesStore.StoredMessage msg) {
this.sessionsStore.moveInFlightToSecondPhaseAckWaiting(this.clientID, messageID, msg);
}
@Override
public boolean isEmptyQueue() {
return this.sessionsStore.queue(clientID).isEmpty();
}
@Override
public void dropQueue() {
LOG.debug("Removing messages of session. CId={}", this.clientID);
this.sessionsStore.dropQueue(this.clientID);
LOG.debug("Messages of the session have been removed. CId={}", this.clientID);
}
@Override
public EnqueuedMessage poll() {
IMessagesStore.StoredMessage msg = this.sessionsStore.queue(clientID).poll();
if (msg == null) {
return null;
}
int messageId = this.inFlightAckWaiting(msg);
return new EnqueuedMessage(msg, messageId);
}
@Override
public void enqueue(IMessagesStore.StoredMessage message) {
this.sessionsStore.queue(this.clientID).add(message);
}
@Override
public IMessagesStore.StoredMessage completeReleasedPublish(int messageID) {
return this.sessionsStore.completeReleasedPublish(clientID, messageID);
}
@Override
public int getPendingPublishMessagesNo() {
return this.sessionsStore.queue(clientID).size();
}
@Override
public int countPubReleaseWaitingPubComplete() {
return this.sessionsStore.countPubReleaseWaitingPubComplete(clientID);
}
@Override
public int getInflightMessagesNo() {
return this.sessionsStore.getInflightMessagesNo(clientID);
}
}
broker/src/main/java/io/moquette/spi/EnqueuedMessage.java
+28
View File
@@ -0,0 +1,28 @@
/*
* Copyright (c) 2012-2017 The original author or authors
* ------------------------------------------------------
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
* and Apache License v2.0 which accompanies this distribution.
*
* The Eclipse Public License is available at
* http://www.eclipse.org/legal/epl-v10.html
*
* The Apache License v2.0 is available at
* http://www.opensource.org/licenses/apache2.0.php
*
* You may elect to redistribute this code under either of these licenses.
*/
package io.moquette.spi;
public class EnqueuedMessage {
public final IMessagesStore.StoredMessage msg;
public final int messageId;
EnqueuedMessage(IMessagesStore.StoredMessage msg, int messageId) {
this.msg = msg;
this.messageId = messageId;
}
}
broker/src/main/java/io/moquette/spi/IMessagesStore.java
-1
View File
@@ -21,7 +21,6 @@ import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.handler.codec.mqtt.MqttQoS;
import java.io.Serializable;
import java.nio.ByteBuffer;
import java.util.Collection;
/**
broker/src/main/java/io/moquette/spi/ISessionsStore.java
+12 -29
View File
@@ -16,6 +16,7 @@
package io.moquette.spi;
import io.moquette.persistence.PersistentSession;
import io.moquette.spi.IMessagesStore.StoredMessage;
import java.util.Collection;
@@ -30,30 +31,27 @@ public interface ISessionsStore {
ISubscriptionsStore subscriptionStore();
void updateCleanStatus(String clientID, boolean cleanSession);
/**
* @param clientID
* the session client ID.
* @return true iff there are subscriptions persisted with clientID
* @return true iff there is a session stored for the client
*/
boolean contains(String clientID);
ClientSession createNewSession(String clientID, boolean cleanSession);
void createNewDurableSession(String clientID);
/**
* @param clientID
* the client owning the session.
* @return the session for the given clientID, null if not found.
*/
ClientSession sessionForClient(String clientID);
void removeDurableSession(String clientId);
void updateCleanStatus(String clientID, boolean cleanSession);
PersistentSession loadSessionByKey(String clientID);
/**
* Returns all the sessions
*
* @return the collection of all stored client sessions.
*/
Collection<ClientSession> getAllSessions();
Collection<PersistentSession> listAllSessions();
StoredMessage inFlightAck(String clientID, int messageID);
@@ -98,7 +96,7 @@ public interface ISessionsStore {
* the message ID that reached the second phase.
* @return the guid of message just acked.
*/
StoredMessage secondPhaseAcknowledged(String clientID, int messageID);
StoredMessage completeReleasedPublish(String clientID, int messageID);
/**
* Returns the number of inflight messages for the given client ID
@@ -108,28 +106,13 @@ public interface ISessionsStore {
*/
int getInflightMessagesNo(String clientID);
/**
* @return the inflight inbound (PUBREL for Qos2) message.
* */
IMessagesStore.StoredMessage inboundInflight(String clientID, int messageID);
void markAsInboundInflight(String clientID, int messageID, StoredMessage msg);
/**
* Returns the size of the session queue for the given client ID
*
* @param clientID target client.
* @return count of enqueued publish messages.
*/
int getPendingPublishMessagesNo(String clientID);
/**
* Returns the number of second-phase ACK pending messages for the given client ID
*
* @param clientID target client.
* @return count of pending in flight publish messages.
*/
int getSecondPhaseAckPendingMessages(String clientID);
int countPubReleaseWaitingPubComplete(String clientID);
void cleanSession(String clientID);
void removeTemporaryQoS2(String clientID);
}
broker/src/main/java/io/moquette/spi/ISubscriptionsStore.java
+12 -49
View File
@@ -3,46 +3,10 @@ package io.moquette.spi;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.spi.impl.subscriptions.Topic;
import java.util.Collection;
import java.util.List;
public interface ISubscriptionsStore {
class ClientTopicCouple {
public final Topic topicFilter;
public final String clientID;
public ClientTopicCouple(String clientID, Topic topicFilter) {
this.clientID = clientID;
this.topicFilter = topicFilter;
}
@Override
public boolean equals(Object o) {
if (this == o)
return true;
if (o == null || getClass() != o.getClass())
return false;
ClientTopicCouple that = (ClientTopicCouple) o;
if (topicFilter != null ? !topicFilter.equals(that.topicFilter) : that.topicFilter != null)
return false;
return !(clientID != null ? !clientID.equals(that.clientID) : that.clientID != null);
}
@Override
public int hashCode() {
int result = topicFilter != null ? topicFilter.hashCode() : 0;
result = 31 * result + (clientID != null ? clientID.hashCode() : 0);
return result;
}
@Override
public String toString() {
return "ClientTopicCouple{" + "topicFilter='" + topicFilter + '\'' + ", clientID='" + clientID + '\'' + '}';
}
}
/**
* Add a new subscription to the session
@@ -65,26 +29,25 @@ public interface ISubscriptionsStore {
/**
* Remove all the subscriptions of the session
*
* @param sessionID
* @param clientID
* the client ID
*/
void wipeSubscriptions(String sessionID);
void wipeSubscriptions(String clientID);
/**
* Return all topic filters to recreate the subscription tree.
*/
List<ClientTopicCouple> listAllSubscriptions();
List<Subscription> listAllSubscriptions();
/**
* @param couple
* the subscription descriptor.
* Load from storage all the subscriptions of the specified client.
* */
Collection<Subscription> listClientSubscriptions(String clientID);
/**
* @param subcription
* the subscription to reaload from storage.
* @return the subscription stored by clientID and topicFilter, if any else null;
*/
Subscription getSubscription(ClientTopicCouple couple);
/*
* @return all subscriptions stored.
*/
List<Subscription> getSubscriptions();
Subscription reload(Subscription subcription);
}
broker/src/main/java/io/moquette/spi/TransientSession.java
+185
View File
@@ -0,0 +1,185 @@
/*
* Copyright (c) 2012-2017 The original author or authors
* ------------------------------------------------------
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
* and Apache License v2.0 which accompanies this distribution.
*
* The Eclipse Public License is available at
* http://www.eclipse.org/legal/epl-v10.html
*
* The Apache License v2.0 is available at
* http://www.opensource.org/licenses/apache2.0.php
*
* You may elect to redistribute this code under either of these licenses.
*/
package io.moquette.spi;
import io.moquette.spi.IMessagesStore.StoredMessage;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.spi.impl.subscriptions.Topic;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger;
public class TransientSession extends ClientSession {
private static final Logger LOG = LoggerFactory.getLogger(TransientSession.class);
private Queue<StoredMessage> messagesQueue = new LinkedList<>();
private final AtomicInteger packetGenerator = new AtomicInteger(1);
private final ConcurrentMap<Integer, StoredMessage> inboundInflightMap = new ConcurrentHashMap<>();
private final ConcurrentMap<Integer, StoredMessage> outboundInflightMap = new ConcurrentHashMap<>();
private final Map<Integer, StoredMessage> secondPhaseStore = new ConcurrentHashMap<>();
public TransientSession(String clientID) {
super(clientID);
}
@Override
public void disconnect() {}
@Override
public boolean isCleanSession() {
return true;
}
@Override
public boolean subscribe(Subscription newSubscription) {
LOG.info("Adding new subscription. CId={}, topics={}, qos={}", newSubscription.getClientId(),
newSubscription.getTopicFilter(), newSubscription.getRequestedQos());
boolean validTopic = newSubscription.getTopicFilter().isValid();
if (!validTopic) {
LOG.warn("The topic filter is not valid. CId={}, topics={}", newSubscription.getClientId(),
newSubscription.getTopicFilter());
// send SUBACK with 0x80 for this topic filter
return false;
}
final Optional<Subscription> mathingExisting = this.subscriptions.stream()
.filter(s -> s.equals(newSubscription))
.findFirst();
if (mathingExisting.isPresent()) {
Subscription existingSub = mathingExisting.get();
if (existingSub.qosLessThan(newSubscription)) {
LOG.info("Subscription already existed with a lower QoS value. It will be updated. CId={}, " +
"topics={}, existingQos={}, newQos={}", newSubscription.getClientId(),
newSubscription.getTopicFilter(), existingSub.getRequestedQos(), newSubscription.getRequestedQos());
subscriptions.remove(newSubscription);
}
}
subscriptions.add(newSubscription);
return true;
}
@Override
public void unsubscribeFrom(Topic topicFilter) {
LOG.info("Removing subscription. CId={}, topics={}", clientID, topicFilter);
Set<Subscription> subscriptionsToRemove = new HashSet<>();
for (Subscription sub : this.subscriptions) {
if (sub.getTopicFilter().equals(topicFilter)) {
subscriptionsToRemove.add(sub);
}
}
subscriptions.removeAll(subscriptionsToRemove);
}
@Override
public boolean isEmptyQueue() {
return this.messagesQueue.isEmpty();
}
@Override
public void enqueue(StoredMessage message) {
this.messagesQueue.offer(message);
}
@Override
public EnqueuedMessage poll() {
IMessagesStore.StoredMessage msg = this.messagesQueue.poll();
if (msg == null) {
return null;
}
int messageId = this.inFlightAckWaiting(msg);
return new EnqueuedMessage(msg, messageId);
}
@Override
public void dropQueue() {
LOG.debug("Removing messages of session. CId={}", this.clientID);
this.messagesQueue = null;
LOG.debug("Messages of the session have been removed. CId={}", this.clientID);
}
@Override
public void cleanSession() {
LOG.info("Transient sessiom, wiping existing subscriptions. ClientId={}", clientID);
this.subscriptions.clear();
//TODO
// LOG.info("Removing queues. ClientId={}", clientID);
// m_sessionsStore.dropQueue(clientID);
}
@Override
protected int nextPacketId() {
return this.packetGenerator.getAndIncrement();
}
@Override
public int inFlightAckWaiting(StoredMessage msg) {
LOG.debug("Adding message to inflight zone. CId={}", clientID);
int messageId = this.nextPacketId();
this.outboundInflightMap.put(messageId, msg);
return messageId;
}
@Override
public IMessagesStore.StoredMessage inFlightAcknowledged(int messageID) {
LOG.debug("Removing message to inflight zone. CId={}, messageID={}", clientID, messageID);
return this.outboundInflightMap.remove(messageID);
}
@Override
public void markAsInboundInflight(int messageID, IMessagesStore.StoredMessage msg) {
this.inboundInflightMap.put(messageID, msg);
}
@Override
public IMessagesStore.StoredMessage inboundInflight(int messageID) {
return this.inboundInflightMap.remove(messageID);
}
@Override
public void moveInFlightToSecondPhaseAckWaiting(int messageID, IMessagesStore.StoredMessage msg) {
this.secondPhaseStore.put(messageID, msg);
this.outboundInflightMap.put(messageID, msg);
}
@Override
public IMessagesStore.StoredMessage completeReleasedPublish(int messageID) {
LOG.info("Acknowledged message in second phase, clientID <{}> messageID {}", clientID, messageID);
return this.secondPhaseStore.remove(messageID);
}
@Override
public int getPendingPublishMessagesNo() {
return this.messagesQueue.size();
}
@Override
public int countPubReleaseWaitingPubComplete() {
return this.outboundInflightMap.size();
}
@Override
public int getInflightMessagesNo() {
return this.inboundInflightMap.size();
}
}
broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java
+13 -8
View File
@@ -17,6 +17,7 @@
package io.moquette.spi.impl;
import io.moquette.spi.ClientSession;
import io.moquette.spi.EnqueuedMessage;
import io.moquette.spi.IMessagesStore;
import io.netty.buffer.ByteBuf;
import io.netty.handler.codec.mqtt.*;
@@ -24,7 +25,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Collection;
import java.util.Queue;
class InternalRepublisher {
@@ -53,17 +53,22 @@ class InternalRepublisher {
}
}
void publishStored(ClientSession clientSession, Queue<IMessagesStore.StoredMessage> publishedEvents) {
IMessagesStore.StoredMessage pubEvt;
while ((pubEvt = publishedEvents.poll()) != null) {
void publishStored(ClientSession clientSession) {
if (clientSession.isEmptyQueue()) {
LOG.info("There are no stored publish events to CId={}", clientSession.clientID);
return;
}
EnqueuedMessage pubEvt;
while ((pubEvt = clientSession.poll()) != null) {
// 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);
MqttPublishMessage publishMsg = notRetainedPublish(pubEvt);
pubEvt.msg.getGuid(), pubEvt.msg.getTopic());
MqttPublishMessage publishMsg = notRetainedPublish(pubEvt.msg);
// set the PacketIdentifier only for QoS > 0
if (publishMsg.fixedHeader().qosLevel() != MqttQoS.AT_MOST_ONCE) {
publishMsg = notRetainedPublish(pubEvt, messageId);
publishMsg = notRetainedPublish(pubEvt.msg, pubEvt.messageId);
}
this.messageSender.sendPublish(clientSession, publishMsg);
}
broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java
+8 -6
View File
@@ -19,7 +19,6 @@ package io.moquette.spi.impl;
import io.moquette.server.ConnectionDescriptorStore;
import io.moquette.spi.ClientSession;
import io.moquette.spi.IMessagesStore;
import io.moquette.spi.ISessionsStore;
import io.moquette.spi.impl.subscriptions.ISubscriptionsDirectory;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.spi.impl.subscriptions.Topic;
@@ -27,23 +26,26 @@ import io.netty.buffer.ByteBuf;
import io.netty.handler.codec.mqtt.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.List;
import static io.moquette.spi.impl.ProtocolProcessor.lowerQosToTheSubscriptionDesired;
class MessagesPublisher {
private static final Logger LOG = LoggerFactory.getLogger(MessagesPublisher.class);
private final ConnectionDescriptorStore connectionDescriptors;
private final ISessionsStore m_sessionsStore;
private final PersistentQueueMessageSender messageSender;
private final ISubscriptionsDirectory subscriptions;
private SessionsRepository sessionsRepository;
public MessagesPublisher(ConnectionDescriptorStore connectionDescriptors, ISessionsStore sessionsStore,
PersistentQueueMessageSender messageSender, ISubscriptionsDirectory subscriptions) {
public MessagesPublisher(ConnectionDescriptorStore connectionDescriptors,
PersistentQueueMessageSender messageSender, ISubscriptionsDirectory subscriptions,
SessionsRepository sessionsRepository) {
this.connectionDescriptors = connectionDescriptors;
this.m_sessionsStore = sessionsStore;
this.messageSender = messageSender;
this.subscriptions = subscriptions;
this.sessionsRepository = sessionsRepository;
}
static MqttPublishMessage notRetainedPublish(String topic, MqttQoS qos, ByteBuf message) {
@@ -77,7 +79,7 @@ class MessagesPublisher {
for (final Subscription sub : topicMatchingSubscriptions) {
MqttQoS qos = lowerQosToTheSubscriptionDesired(sub, publishingQos);
ClientSession targetSession = m_sessionsStore.sessionForClient(sub.getClientId());
ClientSession targetSession = this.sessionsRepository.sessionForClient(sub.getClientId());
boolean targetIsActive = this.connectionDescriptors.isConnected(sub.getClientId());
//TODO move all this logic into messageSender, which puts into the flightZone only the messages that pull out of the queue.
broker/src/main/java/io/moquette/spi/impl/PersistentQueueMessageSender.java
+2 -2
View File
@@ -51,11 +51,11 @@ class PersistentQueueMessageSender {
if (!messageDelivered) {
if (qos != AT_MOST_ONCE && !clientsession.isCleanSession()) {
LOG.warn("PUBLISH message could not be delivered. It will be stored. MessageId={}, CId={}, topic={}, "
+ "qos={}, cleanSession={}", messageId, clientId, topicName, qos, false);
+ "qos={}, removeTemporaryQoS2={}", messageId, clientId, topicName, qos, false);
clientsession.enqueue(asStoredMessage(pubMessage));
} else {
LOG.warn("PUBLISH message could not be delivered. It will be discarded. MessageId={}, CId={}, topic={}, " +
"qos={}, cleanSession={}", messageId, clientId, topicName, qos, true);
"qos={}, removeTemporaryQoS2={}", messageId, clientId, topicName, qos, true);
}
}
}
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+63 -104
View File
@@ -125,8 +125,8 @@ public class ProtocolProcessor {
private static final Logger LOG = LoggerFactory.getLogger(ProtocolProcessor.class);
protected ConnectionDescriptorStore connectionDescriptors;
protected ConcurrentMap<RunningSubscription, SubscriptionState> subscriptionInCourse;
private ConnectionDescriptorStore connectionDescriptors;
private ConcurrentMap<RunningSubscription, SubscriptionState> subscriptionInCourse;
private ISubscriptionsDirectory subscriptions;
private ISubscriptionsStore subscriptionStore;
@@ -146,6 +146,7 @@ public class ProtocolProcessor {
private Qos2PublishHandler qos2PublishHandler;
private MessagesPublisher messagesPublisher;
private InternalRepublisher internalRepublisher;
SessionsRepository sessionsRepository;
// maps clientID to Will testament, if specified on CONNECT
private ConcurrentMap<String, WillMessage> m_willStore = new ConcurrentHashMap<>();
@@ -155,23 +156,16 @@ public class ProtocolProcessor {
public void init(ISubscriptionsDirectory subscriptions, IMessagesStore storageService, ISessionsStore sessionsStore,
IAuthenticator authenticator, boolean allowAnonymous, IAuthorizator authorizator,
BrokerInterceptor interceptor) {
BrokerInterceptor interceptor, SessionsRepository sessionsRepository) {
init(subscriptions, storageService, sessionsStore, authenticator, allowAnonymous, false, authorizator,
interceptor, null);
interceptor, sessionsRepository);
}
public void init(ISubscriptionsDirectory 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(ISubscriptionsDirectory 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);
IAuthorizator authorizator, BrokerInterceptor interceptor, SessionsRepository sessionsRepository) {
init(new ConnectionDescriptorStore(sessionsRepository), subscriptions, storageService, sessionsStore, authenticator,
allowAnonymous, allowZeroByteClientId, authorizator, interceptor, sessionsRepository);
}
/**
@@ -193,9 +187,9 @@ public class ProtocolProcessor {
* to notify events to an intercept handler
*/
void init(ConnectionDescriptorStore connectionDescriptors, ISubscriptionsDirectory subscriptions,
IMessagesStore storageService, ISessionsStore sessionsStore, IAuthenticator authenticator,
boolean allowAnonymous, boolean allowZeroByteClientId, IAuthorizator authorizator,
BrokerInterceptor interceptor, String serverPort) {
IMessagesStore storageService, ISessionsStore sessionsStore, IAuthenticator authenticator,
boolean allowAnonymous, boolean allowZeroByteClientId, IAuthorizator authorizator,
BrokerInterceptor interceptor, SessionsRepository sessionsRepository) {
LOG.info("Initializing MQTT protocol processor...");
this.connectionDescriptors = connectionDescriptors;
this.subscriptionInCourse = new ConcurrentHashMap<>();
@@ -212,10 +206,12 @@ public class ProtocolProcessor {
m_sessionsStore = sessionsStore;
subscriptionStore = sessionsStore.subscriptionStore();
this.sessionsRepository = sessionsRepository;
LOG.info("Initializing messages publisher...");
final PersistentQueueMessageSender messageSender = new PersistentQueueMessageSender(this.connectionDescriptors);
this.messagesPublisher = new MessagesPublisher(connectionDescriptors, sessionsStore, messageSender,
subscriptions);
this.messagesPublisher = new MessagesPublisher(connectionDescriptors, messageSender,
subscriptions, this.sessionsRepository);
LOG.info("Initializing QoS publish handlers...");
this.qos0PublishHandler = new Qos0PublishHandler(m_authorizator, m_messagesStore, m_interceptor,
@@ -223,7 +219,7 @@ public class ProtocolProcessor {
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, this.messagesPublisher);
this.connectionDescriptors, this.messagesPublisher, this.sessionsRepository);
LOG.info("Initializing internal republisher...");
this.internalRepublisher = new InternalRepublisher(messageSender);
@@ -244,8 +240,9 @@ public class ProtocolProcessor {
return;
}
final boolean cleanSession = msg.variableHeader().isCleanSession();
if (clientId == null || clientId.length() == 0) {
if (!msg.variableHeader().isCleanSession() || !this.allowZeroByteClientId) {
if (!cleanSession || !this.allowZeroByteClientId) {
MqttConnAckMessage badId = connAck(CONNECTION_REFUSED_IDENTIFIER_REJECTED);
channel.writeAndFlush(badId);
@@ -265,8 +262,7 @@ public class ProtocolProcessor {
return;
}
ConnectionDescriptor descriptor = new ConnectionDescriptor(clientId, channel,
msg.variableHeader().isCleanSession());
ConnectionDescriptor descriptor = new ConnectionDescriptor(clientId, channel, cleanSession);
ConnectionDescriptor existing = this.connectionDescriptors.addConnection(descriptor);
if (existing != null) {
LOG.info("Client ID is being used in an existing connection, force to be closed. CId={}", clientId);
@@ -285,11 +281,11 @@ public class ProtocolProcessor {
m_interceptor.notifyClientConnected(msg);
final ClientSession clientSession = createOrLoadClientSession(descriptor, msg, clientId);
if (clientSession == null) {
if (!descriptor.assignState(SENDACK, SESSION_CREATED)) {
channel.close();
return;
}
final ClientSession clientSession = this.sessionsRepository.createOrLoadClientSession(clientId, cleanSession);
if (!republish(descriptor, msg, clientSession)) {
channel.close();
@@ -345,27 +341,30 @@ public class ProtocolProcessor {
}
private boolean sendAck(ConnectionDescriptor descriptor, MqttConnectMessage msg, final String clientId) {
LOG.info("Sending connect ACK. CId={}", clientId);
LOG.info("Sending CONNACK. CId={}", clientId);
final boolean success = descriptor.assignState(DISCONNECTED, SENDACK);
if (!success) {
return false;
}
MqttConnAckMessage okResp;
ClientSession clientSession = m_sessionsStore.sessionForClient(clientId);
ClientSession clientSession = this.sessionsRepository.sessionForClient(clientId);
boolean isSessionAlreadyStored = clientSession != null;
if (!msg.variableHeader().isCleanSession() && isSessionAlreadyStored) {
final boolean msgCleanSessionFlag = msg.variableHeader().isCleanSession();
if (!msgCleanSessionFlag && isSessionAlreadyStored) {
okResp = connAckWithSessionPresent(CONNECTION_ACCEPTED);
} else {
okResp = connAck(CONNECTION_ACCEPTED);
}
if (isSessionAlreadyStored) {
LOG.info("Cleaning session. CId={}", clientId);
clientSession.cleanSession(msg.variableHeader().isCleanSession());
}
descriptor.writeAndFlush(okResp);
LOG.info("The connect ACK has been sent. CId={}", clientId);
LOG.info("CONNACK has been sent. CId={}", clientId);
if (isSessionAlreadyStored && msgCleanSessionFlag) {
for (Subscription existingSub : clientSession.getSubscriptions()) {
this.subscriptions.removeSubscription(existingSub.getTopicFilter(), clientId);
}
}
return true;
}
@@ -373,15 +372,12 @@ public class ProtocolProcessor {
int keepAlive = msg.variableHeader().keepAliveTimeSeconds();
LOG.info("Configuring connection. CId={}", clientId);
NettyUtils.keepAlive(channel, keepAlive);
// session.attr(NettyUtils.ATTR_KEY_CLEANSESSION).set(msg.variableHeader().isCleanSession());
NettyUtils.cleanSession(channel, msg.variableHeader().isCleanSession());
// used to track the client in the subscription and publishing phases.
// session.attr(NettyUtils.ATTR_KEY_CLIENTID).set(msg.getClientID());
NettyUtils.clientID(channel, clientId);
int idleTime = Math.round(keepAlive * 1.5f);
setIdleTime(channel.pipeline(), idleTime);
LOG.debug("The connection has been configured CId={}, keepAlive={}, cleanSession={}, idleTime={}",
LOG.debug("The connection has been configured CId={}, keepAlive={}, removeTemporaryQoS2={}, idleTime={}",
clientId, keepAlive, msg.variableHeader().isCleanSession(), idleTime);
}
@@ -401,25 +397,6 @@ public class ProtocolProcessor {
}
}
private ClientSession createOrLoadClientSession(ConnectionDescriptor descriptor, MqttConnectMessage msg,
String clientId) {
final boolean success = descriptor.assignState(SENDACK, SESSION_CREATED);
if (!success) {
return null;
}
ClientSession clientSession = m_sessionsStore.sessionForClient(clientId);
boolean isSessionAlreadyStored = clientSession != null;
if (!isSessionAlreadyStored) {
clientSession = m_sessionsStore.createNewSession(clientId, msg.variableHeader().isCleanSession());
}
if (msg.variableHeader().isCleanSession()) {
LOG.info("Cleaning session. CId={}", clientId);
clientSession.cleanSession();
}
return clientSession;
}
private boolean republish(ConnectionDescriptor descriptor, MqttConnectMessage msg, ClientSession clientSession) {
final boolean success = descriptor.assignState(SESSION_CREATED, MESSAGES_REPUBLISHED);
if (!success) {
@@ -428,7 +405,8 @@ public class ProtocolProcessor {
if (!msg.variableHeader().isCleanSession()) {
// force the republish of stored QoS1 and QoS2
republishStoredInSession(clientSession);
LOG.info("Republishing stored publish events. CId={}", clientSession.clientID);
this.internalRepublisher.publishStored(clientSession);
}
int flushIntervalMs = 500/* (keepAlive * 1000) / 2 */;
descriptor.setupAutoFlusher(flushIntervalMs);
@@ -447,27 +425,13 @@ public class ProtocolProcessor {
pipeline.addFirst("idleStateHandler", new IdleStateHandler(idleTime, 0, 0));
}
/**
* Republish QoS1 and QoS2 messages stored into the session for the clientID.
*/
private void republishStoredInSession(ClientSession clientSession) {
LOG.info("Republishing stored publish events. CId={}", clientSession.clientID);
Queue<StoredMessage> publishedEvents = clientSession.queue();
if (publishedEvents.isEmpty()) {
LOG.info("There are no stored publish events to CId={}", clientSession.clientID);
return;
}
this.internalRepublisher.publishStored(clientSession, publishedEvents);
}
public void processPubAck(Channel channel, MqttPubAckMessage msg) {
String clientID = NettyUtils.clientID(channel);
int messageID = msg.variableHeader().messageId();
String username = NettyUtils.userName(channel);
LOG.trace("retrieving inflight for messageID <{}>", messageID);
ClientSession targetSession = m_sessionsStore.sessionForClient(clientID);
ClientSession targetSession = this.sessionsRepository.sessionForClient(clientID);
StoredMessage inflightMsg = targetSession.inFlightAcknowledged(messageID);
String topic = inflightMsg.getTopic();
@@ -531,16 +495,12 @@ public class ProtocolProcessor {
final Topic topic = new Topic(msg.variableHeader().topicName());
LOG.info("Sending PUBLISH message. Topic={}, qos={}", topic, qos);
MessageGUID guid = null;
IMessagesStore.StoredMessage toStoreMsg = asStoredMessage(msg);
if (clientId == null || clientId.isEmpty()) {
toStoreMsg.setClientID("BROKER_SELF");
} else {
toStoreMsg.setClientID(clientId);
}
// if (qos == EXACTLY_ONCE) { // QoS2
// guid = m_messagesStore.storePublishForFuture(toStoreMsg);
// }
this.messagesPublisher.publish2Subscribers(toStoreMsg, topic);
if (!msg.fixedHeader().isRetain()) {
@@ -551,10 +511,6 @@ public class ProtocolProcessor {
m_messagesStore.cleanRetained(topic);
return;
}
// if (guid == null) {
// // before wasn't stored
// guid = m_messagesStore.storePublishForFuture(toStoreMsg);
// }
m_messagesStore.storeRetained(topic, toStoreMsg);
}
@@ -598,9 +554,10 @@ public class ProtocolProcessor {
public void processPubRec(Channel channel, MqttMessage msg) {
String clientID = NettyUtils.clientID(channel);
ClientSession targetSession = m_sessionsStore.sessionForClient(clientID);
// remove from the inflight and move to the QoS2 second phase queue
int messageID = messageId(msg);
LOG.debug("Processing PUBREC message. CId={}, messageId={}", clientID, messageID);
ClientSession targetSession = this.sessionsRepository.sessionForClient(clientID);
// remove from the inflight and move to the QoS2 second phase queue
StoredMessage ackedMsg = targetSession.inFlightAcknowledged(messageID);
targetSession.moveInFlightToSecondPhaseAckWaiting(messageID, ackedMsg);
// once received a PUBREC reply with a PUBREL(messageID)
@@ -616,12 +573,13 @@ public class ProtocolProcessor {
int messageID = messageId(msg);
LOG.debug("Processing PUBCOMP message. CId={}, messageId={}", clientID, messageID);
// once received the PUBCOMP then remove the message from the temp memory
ClientSession targetSession = m_sessionsStore.sessionForClient(clientID);
StoredMessage inflightMsg = targetSession.secondPhaseAcknowledged(messageID);
ClientSession targetSession = this.sessionsRepository.sessionForClient(clientID);
StoredMessage inflightMsg = targetSession.completeReleasedPublish(messageID);
String username = NettyUtils.userName(channel);
String topic = inflightMsg.getTopic();
m_interceptor
.notifyMessageAcknowledged(new InterceptAcknowledgedMessage(inflightMsg, topic, username, messageID));
final InterceptAcknowledgedMessage interceptAckMsg = new InterceptAcknowledgedMessage(inflightMsg, topic,
username, messageID);
m_interceptor.notifyMessageAcknowledged(interceptAckMsg);
}
public void processDisconnect(Channel channel) throws InterruptedException {
@@ -671,6 +629,7 @@ public class ProtocolProcessor {
LOG.warn("Another descriptor has been inserted. CId={}", clientID);
return;
}
this.sessionsRepository.disconnect(clientID);
LOG.info("DISCONNECT message has been processed. CId={}", clientID);
}
@@ -684,7 +643,7 @@ public class ProtocolProcessor {
if (descriptor.cleanSession) {
LOG.info("Removing saved subscriptions. CId={}", descriptor.clientID);
subscriptionStore.wipeSubscriptions(clientID);
LOG.info("The saved subscriptions have been removed. CId={}", descriptor.clientID);
LOG.info("Saved subscriptions have been removed. CId={}", descriptor.clientID);
}
return true;
}
@@ -695,10 +654,12 @@ public class ProtocolProcessor {
return false;
}
if (descriptor.cleanSession) {
LOG.debug("Removing messages of session. CId={}", descriptor.clientID);
this.m_sessionsStore.dropQueue(clientID);
LOG.debug("The messages of the session have been removed. CId={}", descriptor.clientID);
final ClientSession clientSession = this.sessionsRepository.sessionForClient(clientID);
if (clientSession.isCleanSession()) {
clientSession.dropQueue();
// LOG.debug("Removing messages of session. CId={}", descriptor.clientID);
// this.m_sessionsStore.dropQueue(clientID);
// LOG.debug("Messages of the session have been removed. CId={}", descriptor.clientID);
}
return true;
}
@@ -747,7 +708,7 @@ public class ProtocolProcessor {
LOG.info("Processing UNSUBSCRIBE message. CId={}, topics={}", clientID, topics);
ClientSession clientSession = m_sessionsStore.sessionForClient(clientID);
ClientSession clientSession = this.sessionsRepository.sessionForClient(clientID);
for (String t : topics) {
Topic topic = new Topic(t);
boolean validTopic = topic.isValid();
@@ -801,7 +762,7 @@ public class ProtocolProcessor {
// save session, persist subscriptions from session
for (Subscription subscription : newSubscriptions) {
subscriptions.add(subscription.asClientTopicCouple());
subscriptions.add(subscription);
}
LOG.info("Sending SUBACK response CId={}, messageId={}", clientID, messageID);
@@ -819,7 +780,7 @@ public class ProtocolProcessor {
}
private List<Subscription> doStoreSubscription(List<MqttTopicSubscription> ackTopics, String clientID) {
ClientSession clientSession = m_sessionsStore.sessionForClient(clientID);
ClientSession clientSession = this.sessionsRepository.sessionForClient(clientID);
List<Subscription> newSubscriptions = new ArrayList<>();
for (MqttTopicSubscription req : ackTopics) {
@@ -827,10 +788,8 @@ public class ProtocolProcessor {
if (req.qualityOfService() == FAILURE) {
continue;
}
Subscription newSubscription = new Subscription(
clientID,
new Topic(req.topicName()),
req.qualityOfService());
Subscription newSubscription = new Subscription(clientID, new Topic(req.topicName()),
req.qualityOfService());
clientSession.subscribe(newSubscription);
newSubscriptions.add(newSubscription);
@@ -848,7 +807,7 @@ public class ProtocolProcessor {
* @return the list of verified topics for the given subscribe message.
*/
private List<MqttTopicSubscription> doVerify(String clientID, String username, MqttSubscribeMessage msg) {
ClientSession clientSession = m_sessionsStore.sessionForClient(clientID);
ClientSession clientSession = this.sessionsRepository.sessionForClient(clientID);
List<MqttTopicSubscription> ackTopics = new ArrayList<>();
final int messageId = messageId(msg);
@@ -903,7 +862,7 @@ public class ProtocolProcessor {
LOG.info("Publishing retained messages CId={}, topics={}, messagesNo={}",
newSubscription.getClientId(), newSubscription.getTopicFilter(), messages.size());
}
ClientSession targetSession = m_sessionsStore.sessionForClient(newSubscription.getClientId());
ClientSession targetSession = this.sessionsRepository.sessionForClient(newSubscription.getClientId());
this.internalRepublisher.publishRetained(targetSession, messages);
// notify the Observables
@@ -912,16 +871,16 @@ public class ProtocolProcessor {
public void notifyChannelWritable(Channel channel) {
String clientID = NettyUtils.clientID(channel);
ClientSession clientSession = m_sessionsStore.sessionForClient(clientID);
ClientSession clientSession = this.sessionsRepository.sessionForClient(clientID);
boolean emptyQueue = false;
while (channel.isWritable() && !emptyQueue) {
StoredMessage msg = clientSession.queue().poll();
EnqueuedMessage msg = clientSession.poll();
if (msg == null) {
emptyQueue = true;
} else {
// recreate a publish from stored publish in queue
MqttPublishMessage pubMsg = createPublishForQos( msg.getTopic(), msg.getQos(), msg.getPayload(),
msg.isRetained(), 0);
MqttPublishMessage pubMsg = createPublishForQos(msg.msg.getTopic(), msg.msg.getQos(), msg.msg.getPayload(),
msg.msg.isRetained(), msg.messageId);
channel.write(pubMsg);
}
}
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java
+6 -7
View File
@@ -30,7 +30,6 @@ import io.moquette.spi.impl.security.*;
import io.moquette.spi.impl.subscriptions.CTrieSubscriptionDirectory;
import io.moquette.spi.impl.subscriptions.ISubscriptionsDirectory;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.spi.impl.subscriptions.SubscriptionsDirectory;
import io.moquette.spi.security.IAuthenticator;
import io.moquette.spi.security.IAuthorizator;
import org.slf4j.Logger;
@@ -98,6 +97,7 @@ public class ProtocolProcessorBootstrapper {
store.initStore();
messagesStore = store.messagesStore();
m_sessionsStore = store.sessionsStore();
SessionsRepository sessionsRepository = new SessionsRepository(this.m_sessionsStore);
this.subscriptionsStore = m_sessionsStore.subscriptionStore();
storeShutdown = new Runnable() {
@@ -121,7 +121,7 @@ public class ProtocolProcessorBootstrapper {
LOG.info("Initializing subscriptions store...");
ISubscriptionsDirectory subscriptions = new CTrieSubscriptionDirectory();
subscriptions.init(m_sessionsStore);
subscriptions.init(sessionsRepository);
LOG.info("Configuring MQTT authenticator...");
String authenticatorClassName = props.getProperty(BrokerConstants.AUTHENTICATOR_CLASS_NAME, "");
@@ -164,7 +164,7 @@ public class ProtocolProcessorBootstrapper {
}
LOG.info("Initializing connection descriptor store...");
connectionDescriptors = new ConnectionDescriptorStore(m_sessionsStore);
connectionDescriptors = new ConnectionDescriptorStore(sessionsRepository);
LOG.info("Initializing MQTT protocol processor...");
boolean allowAnonymous = Boolean
@@ -173,7 +173,7 @@ public class ProtocolProcessorBootstrapper {
.parseBoolean(props.getProperty(BrokerConstants.ALLOW_ZERO_BYTE_CLIENT_ID_PROPERTY_NAME, "false"));
m_processor.init(connectionDescriptors, subscriptions, messagesStore, m_sessionsStore, authenticator,
allowAnonymous, allowZeroByteClientId, authorizator, interceptor,
props.getProperty(BrokerConstants.PORT_PROPERTY_NAME));
sessionsRepository);
return m_processor;
}
@@ -226,8 +226,7 @@ public class ProtocolProcessorBootstrapper {
return null;
} catch (NoSuchMethodException | InvocationTargetException e) {
try {
LOG.info("Invoking default constructor. ClassName={}, interfaceName={}",
className, intrface.getName());
LOG.info("Invoking default constructor. ClassName={}, interfaceName={}", className, intrface.getName());
// fallback to default constructor
instance = this.getClass().getClassLoader()
.loadClass(className)
@@ -248,7 +247,7 @@ public class ProtocolProcessorBootstrapper {
}
public List<Subscription> getSubscriptions() {
return this.subscriptionsStore.getSubscriptions();
return this.subscriptionsStore.listAllSubscriptions();
}
public void shutdown() {
broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java
+1 -1
View File
@@ -93,7 +93,7 @@ class Qos1PublishHandler extends QosPublishHandler {
throw new RuntimeException("Internal bad error, found connectionDescriptors to null while it should " +
"be initialized, somewhere it's overwritten!!");
}
LOG.debug("clientIDs are {}", connectionDescriptors);
LOG.trace("connected clientIDs are {}", connectionDescriptors.getConnectedClientIds());
if (!connectionDescriptors.isConnected(clientId)) {
throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client %s in cache %s",
clientId, connectionDescriptors));
broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java
+6 -7
View File
@@ -43,20 +43,20 @@ class Qos2PublishHandler extends QosPublishHandler {
private final IMessagesStore m_messagesStore;
private final BrokerInterceptor m_interceptor;
private final ConnectionDescriptorStore connectionDescriptors;
private final ISessionsStore m_sessionsStore;
private final MessagesPublisher publisher;
private final SessionsRepository sessionsRepository;
public Qos2PublishHandler(IAuthorizator authorizator, ISubscriptionsDirectory subscriptions,
IMessagesStore messagesStore, BrokerInterceptor interceptor,
ConnectionDescriptorStore connectionDescriptors, ISessionsStore sessionsStore,
MessagesPublisher messagesPublisher) {
ConnectionDescriptorStore connectionDescriptors,
MessagesPublisher messagesPublisher, SessionsRepository sessionsRepository) {
super(authorizator);
this.subscriptions = subscriptions;
this.m_messagesStore = messagesStore;
this.m_interceptor = interceptor;
this.connectionDescriptors = connectionDescriptors;
this.m_sessionsStore = sessionsStore;
this.publisher = messagesPublisher;
this.sessionsRepository = sessionsRepository;
}
void receivedPublishQos2(Channel channel, MqttPublishMessage msg) {
@@ -78,8 +78,7 @@ class Qos2PublishHandler extends QosPublishHandler {
LOG.trace("payload={}, subs Tree={}", payload2Str(toStoreMsg.getPayload()), subscriptions.dumpTree());
}
m_sessionsStore.sessionForClient(clientID)
.markAsInboundInflight(messageID, toStoreMsg);
this.sessionsRepository.sessionForClient(clientID).markAsInboundInflight(messageID, toStoreMsg);
sendPubRec(clientID, messageID);
@@ -105,7 +104,7 @@ class Qos2PublishHandler extends QosPublishHandler {
String clientID = NettyUtils.clientID(channel);
int messageID = messageId(msg);
LOG.info("Processing PUBREL message. CId={}, messageId={}", clientID, messageID);
ClientSession targetSession = m_sessionsStore.sessionForClient(clientID);
ClientSession targetSession = this.sessionsRepository.sessionForClient(clientID);
IMessagesStore.StoredMessage evt = targetSession.inboundInflight(messageID);
if (evt == null) {
LOG.warn("Can't find inbound inflight message for CId={}, messageId={}", clientID, messageID);
broker/src/main/java/io/moquette/spi/impl/SessionsRepository.java
+126
View File
@@ -0,0 +1,126 @@
/*
* Copyright (c) 2012-2017 The original author or authors
* ------------------------------------------------------
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
* and Apache License v2.0 which accompanies this distribution.
*
* The Eclipse Public License is available at
* http://www.eclipse.org/legal/epl-v10.html
*
* The Apache License v2.0 is available at
* http://www.opensource.org/licenses/apache2.0.php
*
* You may elect to redistribute this code under either of these licenses.
*/
package io.moquette.spi.impl;
import io.moquette.persistence.PersistentSession;
import io.moquette.spi.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
public class SessionsRepository {
private static final Logger LOG = LoggerFactory.getLogger(SessionsRepository.class);
private final ISessionsStore sessions;
private ISubscriptionsStore subscriptionsStore;
private final Map<String, ClientSession> sessionsCache = new ConcurrentHashMap<>();
public SessionsRepository(ISessionsStore sessionsStore) {
this.sessions = sessionsStore;
this.subscriptionsStore = sessionsStore.subscriptionStore();
}
public ClientSession sessionForClient(String clientID) {
LOG.debug("Retrieving session. CId={}", clientID);
if (this.sessionsCache.containsKey(clientID)) {
return this.sessionsCache.get(clientID);
}
if (!this.sessions.contains(clientID)) {
LOG.warn("Session does not exist. CId={}", clientID);
return null;
}
PersistentSession storedSession = this.sessions.loadSessionByKey(clientID);
return newClientSessionAndCacheIt(storedSession.clientID, storedSession.cleanSession);
}
private ClientSession newClientSessionAndCacheIt(String clientID, boolean cleanSession) {
ClientSession session;
if (cleanSession) {
session = new TransientSession(clientID);
} else {
DurableSession durableSession = new DurableSession(clientID, this.sessions, this.subscriptionsStore);
durableSession.reloadAllSubscriptionsFromStore();
session = durableSession;
}
this.sessionsCache.put(clientID, session);
return session;
}
public ClientSession createNewSession(String clientID, boolean cleanSession) {
if (sessions.contains(clientID)) {
LOG.error("Unable to create a new session: the client ID is already in use. CId={}, cleanSession={}",
clientID, cleanSession);
throw new IllegalArgumentException("Can't create a session with the ID of an already existing" + clientID);
}
LOG.debug("Creating new session. CId={}, cleanSession={}", clientID, cleanSession);
if (!cleanSession) {
sessions.createNewDurableSession(clientID);
}
return newClientSessionAndCacheIt(clientID, cleanSession);
}
public Collection<ClientSession> getAllSessions() {
Collection<ClientSession> result = new ArrayList<>();
for (PersistentSession persistentSession : this.sessions.listAllSessions()) {
result.add(sessionForClient(persistentSession.clientID));
}
return result;
}
private void updateCleanStatus(String clientID, boolean cleanSession) {
LOG.info("Updating cleanSession flag. CId={}, cleanSession={}", clientID, cleanSession);
this.sessions.updateCleanStatus(clientID, cleanSession);
}
ClientSession createOrLoadClientSession(String clientId, boolean cleanSession) {
ClientSession clientSession = this.sessionForClient(clientId);
if (clientSession == null) {
clientSession = this.createNewSession(clientId, cleanSession);
} else {
//session was already present
if (!clientSession.isCleanSession() && cleanSession) {
//remove from storage & create new transient session
//TODO existing subscription from durable session has to be copied into the new transient?
sessions.removeDurableSession(clientId);
clientSession = this.createNewSession(clientId, true);
} /*else {
clientSession.cleanSession(cleanSession);
this.updateCleanStatus(clientId, cleanSession);
}*/
}
if (cleanSession) {
LOG.info("Cleaning session. CId={}", clientId);
clientSession.cleanSession();
}
return clientSession;
}
public void disconnect(String clientId) {
LOG.debug("Removing session from repository's cache");
ClientSession clientSession = this.sessionForClient(clientId);
if (clientSession == null) {
return;
}
sessionsCache.remove(clientId);
}
}
broker/src/main/java/io/moquette/spi/impl/subscriptions/CNode.java
+15 -28
View File
@@ -15,15 +15,16 @@
*/
package io.moquette.spi.impl.subscriptions;
import io.moquette.spi.ISubscriptionsStore.ClientTopicCouple;
import java.util.*;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
class CNode {
Token token;
private List<INode> children;
Set<ClientTopicCouple> subscriptions = new HashSet<>();
Set<Subscription> subscriptions = new HashSet<>();
//private int subtreeSubscriptions;
@@ -33,7 +34,7 @@ class CNode {
}
//Copy constructor
private CNode(Token token, List<INode> children, Set<ClientTopicCouple> subscriptions) {
private CNode(Token token, List<INode> children, Set<Subscription> subscriptions) {
this.token = token;
this.children = children;
this.subscriptions = subscriptions;
@@ -64,10 +65,7 @@ class CNode {
}
private boolean equals(Token token) {
if (this.token == null) {
return false;
}
return this.token.equals(token);
return this.token != null && this.token.equals(token);
}
CNode copy() {
@@ -79,7 +77,7 @@ class CNode {
}
CNode addSubscription(String clientId, Topic topic) {
this.subscriptions.add(new ClientTopicCouple(clientId, topic));
this.subscriptions.add(new Subscription(clientId, topic));
return this;
}
@@ -87,8 +85,8 @@ class CNode {
* @return true iff the subscriptions contained in this node are owned by clientId
* */
boolean containsOnly(String clientId) {
for (ClientTopicCouple sub : this.subscriptions) {
if (!sub.clientID.equals(clientId)) {
for (Subscription sub : this.subscriptions) {
if (!sub.clientId.equals(clientId)) {
return false;
}
}
@@ -97,8 +95,8 @@ class CNode {
//TODO this is equivalent to negate(containsOnly(clientId))
public boolean contains(String clientId) {
for (ClientTopicCouple sub : this.subscriptions) {
if (sub.clientID.equals(clientId)) {
for (Subscription sub : this.subscriptions) {
if (sub.clientId.equals(clientId)) {
return true;
}
}
@@ -106,23 +104,12 @@ class CNode {
}
void removeSubscriptionsFor(String clientId) {
Set<ClientTopicCouple> toRemove = new HashSet<>();
for (ClientTopicCouple sub : this.subscriptions) {
if (sub.clientID.equals(clientId)) {
Set<Subscription> toRemove = new HashSet<>();
for (Subscription sub : this.subscriptions) {
if (sub.clientId.equals(clientId)) {
toRemove.add(sub);
}
}
this.subscriptions.removeAll(toRemove);
}
// public List<INode> childrenMatching(Token token) {
// List<INode> res = new ArrayList<>();
// for (INode iNode : children) {
// final CNode child = iNode.mainNode();
// if (child.equals(token) || (child.token == Token.MULTI || child.token == Token.SINGLE)) {
// res.add(iNode);
// }
// }
// return res;
// }
}
broker/src/main/java/io/moquette/spi/impl/subscriptions/CTrieSubscriptionDirectory.java
+34 -27
View File
@@ -15,9 +15,8 @@
*/
package io.moquette.spi.impl.subscriptions;
import io.moquette.spi.ISessionsStore;
import io.moquette.spi.ISubscriptionsStore;
import io.moquette.spi.ISubscriptionsStore.ClientTopicCouple;
import io.moquette.spi.ClientSession;
import io.moquette.spi.impl.SessionsRepository;
import io.netty.util.internal.StringUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -25,14 +24,16 @@ import org.slf4j.LoggerFactory;
import java.util.*;
import java.util.concurrent.atomic.AtomicInteger;
import static java.lang.String.format;
public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory {
private static final Logger LOG = LoggerFactory.getLogger(CTrieSubscriptionDirectory.class);
private static final Token ROOT = new Token("root");
protected INode root;
private volatile ISubscriptionsStore subscriptionsStore;
INode root;
private volatile SessionsRepository sessionsRepository;
private interface IVisitor<T> {
@@ -57,10 +58,10 @@ public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory {
}
StringBuilder subScriptionsStr = new StringBuilder(" ~~[");
int counter = 0;
for (ClientTopicCouple couple : node.subscriptions) {
for (Subscription couple : node.subscriptions) {
subScriptionsStr
.append("{filter=").append(couple.topicFilter).append(", ")
.append("client='").append(couple.clientID).append("'}");
.append("client='").append(couple.clientId).append("'}");
counter++;
if (counter < node.subscriptions.size()) {
subScriptionsStr.append(";");
@@ -90,24 +91,24 @@ public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory {
OK, REPEAT
}
public void init(ISessionsStore sessionsStore) {
public void init(SessionsRepository sessionsRepository) {
LOG.info("Initializing CTrie");
final CNode mainNode = new CNode();
mainNode.token = ROOT;
this.root = new INode(mainNode);
LOG.info("Initializing subscriptions store...");
this.subscriptionsStore = sessionsStore.subscriptionStore();
List<ClientTopicCouple> subscriptions = this.subscriptionsStore.listAllSubscriptions();
this.sessionsRepository = sessionsRepository;
// reload any subscriptions persisted
if (LOG.isTraceEnabled()) {
LOG.trace("Reloading all stored subscriptions. SubscriptionTree = {}", dumpTree());
}
for (ClientTopicCouple clientTopic : subscriptions) {
LOG.info("Re-subscribing client to topic CId={}, topicFilter={}", clientTopic.clientID,
clientTopic.topicFilter);
add(clientTopic);
for (ClientSession session : this.sessionsRepository.getAllSessions()) {
for (Subscription subscription : session.getSubscriptions()) {
LOG.info("Re-subscribing client to topic CId={}, topicFilter={}", subscription.clientId,
subscription.topicFilter);
add(subscription);
}
}
if (LOG.isTraceEnabled()) {
LOG.trace("Stored subscriptions have been reloaded. SubscriptionTree = {}", dumpTree());
@@ -145,25 +146,31 @@ public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory {
* @return the list of matching subscriptions, or empty if not matching.
*/
Set<Subscription> match(Topic topic) {
final Set<ClientTopicCouple> matchingSubs = recursiveMatch(topic, this.root);
final Set<Subscription> matchingSubs = recursiveMatch(topic, this.root);
// remove the overlapping subscriptions, selecting ones with greatest qos
Map<String, Subscription> subsForClient = new HashMap<>();
for (ClientTopicCouple matchingCouple : matchingSubs) {
Subscription existingSub = subsForClient.get(matchingCouple.clientID);
Subscription sub = this.subscriptionsStore.getSubscription(matchingCouple);
if (sub == null) {
// if the m_sessionStore hasn't the sub because the client disconnected
for (Subscription matchingSub : matchingSubs) {
Subscription existingSub = subsForClient.get(matchingSub.clientId);
final ClientSession subscribedSession = this.sessionsRepository.sessionForClient(matchingSub.clientId);
if (subscribedSession == null) {
//clean session disconnected
continue;
}
Subscription sub = subscribedSession.findSubscriptionByTopicFilter(matchingSub);
if (sub == null) {
final String excpMesg = format("Target session %s is connected but doesn't anymore subscribed to %s",
matchingSub.clientId, matchingSub);
throw new IllegalStateException(excpMesg);
}
// update the selected subscriptions if not present or if has a greater qos
if (existingSub == null || existingSub.getRequestedQos().value() < sub.getRequestedQos().value()) {
subsForClient.put(matchingCouple.clientID, sub);
if (existingSub == null || existingSub.qosLessThan(sub)) {
subsForClient.put(sub.clientId, sub);
}
}
return new HashSet<>(subsForClient.values());
}
Set<ClientTopicCouple> recursiveMatch(Topic topic, INode inode) {
Set<Subscription> recursiveMatch(Topic topic, INode inode) {
CNode cnode = inode.mainNode();
if (cnode.token == Token.MULTI) {
return cnode.subscriptions;
@@ -179,7 +186,7 @@ public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory {
return Collections.emptySet();
}
Topic remainingTopic = (cnode.token == ROOT) ? topic : topic.exceptHeadToken();
Set<ClientTopicCouple> subscriptions = new HashSet<>();
Set<Subscription> subscriptions = new HashSet<>();
if (remainingTopic.isEmpty()) {
subscriptions.addAll(cnode.subscriptions);
}
@@ -190,10 +197,10 @@ public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory {
}
public void add(ClientTopicCouple newSubscription) {
public void add(Subscription newSubscription) {
Action res;
do {
res = insert(newSubscription.clientID, newSubscription.topicFilter, this.root, newSubscription.topicFilter);
res = insert(newSubscription.clientId, newSubscription.topicFilter, this.root, newSubscription.topicFilter);
} while (res == Action.REPEAT);
}
broker/src/main/java/io/moquette/spi/impl/subscriptions/ISubscriptionsDirectory.java
+3 -4
View File
@@ -15,16 +15,15 @@
*/
package io.moquette.spi.impl.subscriptions;
import io.moquette.spi.ISessionsStore;
import io.moquette.spi.ISubscriptionsStore;
import io.moquette.spi.impl.SessionsRepository;
import java.util.List;
public interface ISubscriptionsDirectory {
void init(ISessionsStore sessionsStore);
void init(SessionsRepository sessionsRepository);
void add(ISubscriptionsStore.ClientTopicCouple newSubscription);
void add(Subscription newSubscription);
void removeSubscription(Topic topic, String clientID);
broker/src/main/java/io/moquette/spi/impl/subscriptions/Subscription.java
+17 -7
View File
@@ -16,8 +16,8 @@
package io.moquette.spi.impl.subscriptions;
import io.moquette.spi.ISubscriptionsStore.ClientTopicCouple;
import io.netty.handler.codec.mqtt.MqttQoS;
import java.io.Serializable;
/**
@@ -26,10 +26,10 @@ import java.io.Serializable;
public final class Subscription implements Serializable {
private static final long serialVersionUID = -3383457629635732794L;
final MqttQoS requestedQos; // max QoS acceptable
private final MqttQoS requestedQos; // max QoS acceptable
final String clientId;
final Topic topicFilter;
final boolean active;
private final boolean active;
public Subscription(String clientId, Topic topicFilter, MqttQoS requestedQos) {
this.requestedQos = requestedQos;
@@ -45,6 +45,16 @@ public final class Subscription implements Serializable {
this.active = orig.active;
}
/**
* Constructor with undefined maximum QoS
* */
public Subscription(String clientId, Topic topicFilter) {
this.requestedQos = null;
this.clientId = clientId;
this.topicFilter = topicFilter;
this.active = true;
}
public String getClientId() {
return clientId;
}
@@ -61,6 +71,10 @@ public final class Subscription implements Serializable {
return active;
}
public boolean qosLessThan(Subscription sub) {
return requestedQos.value() < sub.requestedQos.value();
}
@Override
public boolean equals(Object o) {
if (this == o)
@@ -100,8 +114,4 @@ public final class Subscription implements Serializable {
return null;
}
}
public ClientTopicCouple asClientTopicCouple() {
return new ClientTopicCouple(this.clientId, this.topicFilter);
}
}
broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionsDirectory.java
-243
View File
@@ -1,243 +0,0 @@
/*
* Copyright (c) 2012-2017 The original author or authors
* ------------------------------------------------------
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
* and Apache License v2.0 which accompanies this distribution.
*
* The Eclipse Public License is available at
* http://www.eclipse.org/legal/epl-v10.html
*
* The Apache License v2.0 is available at
* http://www.opensource.org/licenses/apache2.0.php
*
* You may elect to redistribute this code under either of these licenses.
*/
package io.moquette.spi.impl.subscriptions;
import io.moquette.spi.ISessionsStore;
import io.moquette.spi.ISubscriptionsStore;
import io.moquette.spi.ISubscriptionsStore.ClientTopicCouple;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.*;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.atomic.AtomicReference;
/**
* Represents a tree of topics subscriptions.
*/
public class SubscriptionsDirectory implements ISubscriptionsDirectory {
static class NodeCouple {
final TreeNode root;
final TreeNode createdNode;
NodeCouple(TreeNode root, TreeNode createdNode) {
this.root = root;
this.createdNode = createdNode;
}
}
public interface IVisitor<T> {
void visit(TreeNode node, int deep);
T getResult();
}
private class DumpTreeVisitor implements IVisitor<String> {
String s = "";
@Override
public void visit(TreeNode node, int deep) {
StringBuilder subScriptionsStr = new StringBuilder();
String indentTabs = indentTabs(deep);
for (ClientTopicCouple couple : node.m_subscriptions) {
subScriptionsStr.append(indentTabs).append(couple.toString()).append("\n");
}
s += node.getToken() == null ? "" : node.getToken().toString();
s += "\n" + (node.m_subscriptions.isEmpty() ? indentTabs : "")
+ subScriptionsStr /* + "\n" */;
}
private String indentTabs(int deep) {
StringBuilder s = new StringBuilder();
for (int i = 0; i < deep; i++) {
s.append("\t");
// s += "--";
}
return s.toString();
}
@Override
public String getResult() {
return s;
}
}
private AtomicReference<TreeNode> subscriptions = new AtomicReference<>(new TreeNode());
private static final Logger LOG = LoggerFactory.getLogger(SubscriptionsDirectory.class);
private volatile ISubscriptionsStore subscriptionsStore;
/**
* Initialize the subscription tree with the list of subscriptions. Maintained for compatibility
* reasons.
*
* @param sessionsStore
* to be used as backing store from the subscription store.
*/
@Override
public void init(ISessionsStore sessionsStore) {
LOG.info("Initializing subscriptions store...");
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());
}
for (ClientTopicCouple clientTopic : subscriptions) {
LOG.info("Re-subscribing client to topic CId={}, topicFilter={}", clientTopic.clientID,
clientTopic.topicFilter);
add(clientTopic);
}
if (LOG.isTraceEnabled()) {
LOG.trace("Stored subscriptions have been reloaded. SubscriptionTree = {}", dumpTree());
}
}
@Override
public void add(ClientTopicCouple newSubscription) {
/*
* The topic filters have already been validated at the ProtocolProcessor. We can assume
* they are valid.
*/
TreeNode oldRoot;
NodeCouple couple;
do {
oldRoot = subscriptions.get();
couple = recreatePath(newSubscription.topicFilter, oldRoot);
couple.createdNode.addSubscription(newSubscription); //createdNode could be null?
couple.root.recalculateSubscriptionsSize();
//spin lock repeating till we can, swap root, if can't swap just re-do the operation
} while (!subscriptions.compareAndSet(oldRoot, couple.root));
LOG.debug("A subscription has been added. Root = {}, oldRoot = {}.", couple.root, oldRoot);
}
NodeCouple recreatePath(Topic topic, final TreeNode oldRoot) {
final TreeNode newRoot = oldRoot.copy();
TreeNode parent = newRoot;
TreeNode current = newRoot;
for (Token token : topic.getTokens()) {
TreeNode matchingChildren;
// check if a children with the same token already exists
if ((matchingChildren = current.childWithToken(token)) != null) {
// copy the traversed node
current = matchingChildren.copy();
// update the child just added in the children list
parent.updateChild(matchingChildren, current);
parent = current;
} else {
// create a new node for the newly inserted token
matchingChildren = new TreeNode();
matchingChildren.setToken(token);
current.addChild(matchingChildren);
current = matchingChildren;
}
}
return new NodeCouple(newRoot, current);
}
@Override
public void removeSubscription(Topic topic, String clientID) {
/*
* The topic filters have already been validated at the ProtocolProcessor. We can assume
* they are valid.
*/
TreeNode oldRoot;
NodeCouple couple;
do {
oldRoot = subscriptions.get();
couple = recreatePath(topic, oldRoot);
couple.createdNode.remove(new ClientTopicCouple(clientID, topic));
couple.root.recalculateSubscriptionsSize();
//spin lock repeating till we can, swap root, if can't swap just re-do the operation
} while (!subscriptions.compareAndSet(oldRoot, couple.root));
}
// /**
// * Visit the topics tree to remove matching subscriptions with clientID. It's a mutating
// * structure operation so create a new subscription tree (partial or total).
// *
// * @param clientID
// * the client ID to remove.
// */
// void removeForClient(String clientID) {
// TreeNode oldRoot;
// TreeNode newRoot;
// do {
// oldRoot = subscriptions.get();
// newRoot = oldRoot.removeClientSubscriptions(clientID);
// // spin lock repeating till we can, swap root, if can't swap just re-do the operation
// } while (!subscriptions.compareAndSet(oldRoot, newRoot));
// }
/**
* Given a topic string return the clients subscriptions that matches it. Topic string can't
* contain character # and + because they are reserved to listeners subscriptions, and not topic
* publishing.
*
* @param topic
* to use fo searching matching subscriptions.
* @return the list of matching subscriptions, or empty if not matching.
*/
@Override
public List<Subscription> matches(Topic topic) {
Queue<Token> tokenQueue = new LinkedBlockingDeque<>(topic.getTokens());
List<ClientTopicCouple> matchingSubs = new ArrayList<>();
subscriptions.get().matches(tokenQueue, matchingSubs);
// remove the overlapping subscriptions, selecting ones with greatest qos
Map<String, Subscription> subsForClient = new HashMap<>();
for (ClientTopicCouple matchingCouple : matchingSubs) {
Subscription existingSub = subsForClient.get(matchingCouple.clientID);
Subscription sub = this.subscriptionsStore.getSubscription(matchingCouple);
if (sub == null) {
// if the m_sessionStore hasn't the sub because the client disconnected
continue;
}
// update the selected subscriptions if not present or if has a greater qos
if (existingSub == null || existingSub.getRequestedQos().value() < sub.getRequestedQos().value()) {
subsForClient.put(matchingCouple.clientID, sub);
}
}
return new ArrayList<>(subsForClient.values());
}
@Override
public int size() {
return subscriptions.get().size();
}
@Override
public String dumpTree() {
DumpTreeVisitor visitor = new DumpTreeVisitor();
bfsVisit(subscriptions.get(), visitor, 0);
return visitor.getResult();
}
private void bfsVisit(TreeNode node, IVisitor<?> visitor, int deep) {
if (node == null) {
return;
}
visitor.visit(node, deep);
for (TreeNode child : node.m_children) {
bfsVisit(child, visitor, ++deep);
}
}
}
broker/src/main/java/io/moquette/spi/impl/subscriptions/TreeNode.java
-180
View File
@@ -1,180 +0,0 @@
/*
* Copyright (c) 2012-2017 The original author or authors
* ------------------------------------------------------
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
* and Apache License v2.0 which accompanies this distribution.
*
* The Eclipse Public License is available at
* http://www.eclipse.org/legal/epl-v10.html
*
* The Apache License v2.0 is available at
* http://www.opensource.org/licenses/apache2.0.php
*
* You may elect to redistribute this code under either of these licenses.
*/
package io.moquette.spi.impl.subscriptions;
import io.moquette.spi.ISubscriptionsStore.ClientTopicCouple;
import java.util.*;
import java.util.concurrent.LinkedBlockingQueue;
class TreeNode {
Token m_token;
List<TreeNode> m_children = new ArrayList<>();
Set<ClientTopicCouple> m_subscriptions = new HashSet<>();
private int subtreeSubscriptions;
TreeNode() {
}
Token getToken() {
return m_token;
}
void setToken(Token topic) {
this.m_token = topic;
}
void addSubscription(ClientTopicCouple s) {
m_subscriptions.add(s);
this.subtreeSubscriptions++;
}
void addChild(TreeNode child) {
m_children.add(child);
this.subtreeSubscriptions += child.subtreeSubscriptions;
}
/**
* Creates a shallow copy of the current node. Copy the token and the children.
*/
TreeNode copy() {
final TreeNode copy = new TreeNode();
copy.m_children = new ArrayList<>(m_children);
copy.m_subscriptions = new HashSet<>(m_subscriptions);
copy.m_token = m_token;
copy.subtreeSubscriptions = this.subtreeSubscriptions;
return copy;
}
/**
* Search for children that has the specified token, if not found return null;
*/
TreeNode childWithToken(Token token) {
for (TreeNode child : m_children) {
if (child.getToken().equals(token)) {
return child;
}
}
return null;
}
void updateChild(TreeNode oldChild, TreeNode newChild) {
m_children.remove(oldChild);
m_children.add(newChild);
this.subtreeSubscriptions += newChild.subtreeSubscriptions - oldChild.subtreeSubscriptions;
}
Collection<ClientTopicCouple> subscriptions() {
return m_subscriptions;
}
public void remove(ClientTopicCouple clientTopicCouple) {
m_subscriptions.remove(clientTopicCouple);
this.subtreeSubscriptions--;
}
// TODO smell a query method that return the result modifing the parameter (matchingSubs)
void matches(Queue<Token> tokens, List<ClientTopicCouple> matchingSubs) {
Token t = tokens.poll();
// check if t is null <=> tokens finished
if (t == null) {
matchingSubs.addAll(m_subscriptions);
// check if it has got a MULTI child and add its subscriptions
for (TreeNode n : m_children) {
if (n.getToken() == Token.MULTI || n.getToken() == Token.SINGLE) {
matchingSubs.addAll(n.subscriptions());
}
}
return;
}
// we are on MULTI, than add subscriptions and return
if (m_token == Token.MULTI) {
matchingSubs.addAll(m_subscriptions);
return;
}
for (TreeNode n : m_children) {
if (n.getToken().match(t)) {
// Create a copy of token, else if navigate 2 sibling it
// consumes 2 elements on the queue instead of one
n.matches(new LinkedBlockingQueue<>(tokens), matchingSubs);
// TODO don't create a copy n.matches(tokens, matchingSubs);
}
}
}
/**
* Return the number of registered subscriptions
*/
int size() {
return this.subtreeSubscriptions;
}
/**
* Create a copied subtree rooted on this node but purged of clientID's subscriptions.
*/
TreeNode removeClientSubscriptions(String clientID) {
// collect what to delete and then delete to avoid ConcurrentModification
TreeNode newSubRoot = this.copy();
remoteSubscriptions(clientID, newSubRoot);
removeSubscriptionFromChildren(clientID, newSubRoot);
return newSubRoot;
}
private void removeSubscriptionFromChildren(String clientID, TreeNode newSubRoot) {
int newSubtreeSubscriptions = 0;
//go deep
List<TreeNode> newChildren = new ArrayList<>(newSubRoot.m_children.size());
for (TreeNode child : newSubRoot.m_children) {
final TreeNode purgedSubtree = child.removeClientSubscriptions(clientID);
if (purgedSubtree.size() != 0) {
newSubtreeSubscriptions += purgedSubtree.size();
newChildren.add(purgedSubtree);
}
}
newSubRoot.m_children = newChildren;
newSubRoot.subtreeSubscriptions += newSubtreeSubscriptions;
}
private void remoteSubscriptions(String clientID, TreeNode newSubRoot) {
List<ClientTopicCouple> subsToRemove = new ArrayList<>();
for (ClientTopicCouple s : newSubRoot.m_subscriptions) {
if (s.clientID.equals(clientID)) {
subsToRemove.add(s);
}
}
for (ClientTopicCouple s : subsToRemove) {
newSubRoot.m_subscriptions.remove(s);
}
newSubRoot.subtreeSubscriptions = newSubRoot.m_subscriptions.size();
}
int recalculateSubscriptionsSize() {
int res = m_subscriptions.size();
for (TreeNode child : m_children) {
res += child.recalculateSubscriptionsSize();
}
this.subtreeSubscriptions = res;
return res;
}
}