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

Refactored the disconenct and clean of session to use new ClientSession

Browse Source
This commit is contained in:
andsel
2015-11-08 18:42:03 +01:00
parent 635c87389d
commit 19dac6e32b
21 changed files with 239 additions and 264 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/eclipse/moquette/server/netty/NettyChannel.java
+1 -1
View File
@@ -71,6 +71,6 @@ public class NettyChannel implements ServerChannel {
@Override
public String toString() {
String clientID = (String) getAttribute(ATTR_KEY_CLIENTID);
return "session [clientID: "+ clientID +"]" + super.toString();
return "session [clientID: "+ clientID +"]";
}
}
broker/src/main/java/org/eclipse/moquette/spi/ClientSession.java
+39 -2
View File
@@ -22,7 +22,6 @@ import org.eclipse.moquette.spi.impl.subscriptions.SubscriptionsStore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
@@ -42,10 +41,16 @@ public class ClientSession {
private Set<Subscription> subscriptions = new HashSet<>();
public ClientSession(String clientID, IMessagesStore messagesStore, ISessionsStore sessionsStore) {
public final boolean cleanSession;
private boolean active = false;
public ClientSession(String clientID, IMessagesStore messagesStore, ISessionsStore sessionsStore,
boolean cleanSession) {
this.clientID = clientID;
this.messagesStore = messagesStore;
this.m_sessionsStore = sessionsStore;
this.cleanSession = cleanSession;
}
//List of client's subscriptions
//list of messages not acknowledged by client
@@ -96,4 +101,36 @@ public class ClientSession {
}
subscriptions.removeAll(subscriptionsToRemove);
}
public void disconnect() {
if (this.cleanSession) {
//cleanup topic subscriptions
cleanSession();
}
//deactivate the session
deactivate();
}
public void cleanSession() {
LOG.info("cleaning old saved subscriptions for client <{}>", this.clientID);
m_sessionsStore.wipeSubscriptions(this.clientID);
//remove also the messages stored of type QoS1/2
messagesStore.dropMessagesInSession(this.clientID);
}
public void activate() {
this.active = true;
this.m_sessionsStore.activate(this.clientID);
}
public void deactivate() {
this.active = false;
this.m_sessionsStore.deactivate(this.clientID);
}
public boolean isActive() {
return this.active;
}
}
broker/src/main/java/org/eclipse/moquette/spi/ISessionsStore.java
+5 -1
View File
@@ -54,10 +54,14 @@ public interface ISessionsStore {
*/
boolean contains(String clientID);
ClientSession createNewSession(String clientID);
ClientSession createNewSession(String clientID, boolean cleanSession);
/**
* @return the session for the given clientID, null if not found.
* */
ClientSession sessionForClient(String clientID);
void activate(String clientID);
void deactivate(String clientID);
}
broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java
+21 -41
View File
@@ -93,7 +93,7 @@ public class ProtocolProcessor {
private static final Logger LOG = LoggerFactory.getLogger(ProtocolProcessor.class);
private Map<String, ConnectionDescriptor> m_clientIDs = new ConcurrentHashMap<>();
private Map<String, ConnectionDescriptor> m_clientIDs;
private SubscriptionsStore subscriptions;
private boolean allowAnonymous;
private IAuthorizator m_authorizator;
@@ -123,6 +123,7 @@ public class ProtocolProcessor {
IAuthenticator authenticator,
boolean allowAnonymous, IAuthorizator authorizator, BrokerInterceptor interceptor) {
//m_clientIDs = clientIDs;
this.m_clientIDs = new ConcurrentHashMap<>();
this.m_interceptor = interceptor;
this.subscriptions = subscriptions;
this.allowAnonymous = allowAnonymous;
@@ -176,12 +177,8 @@ public class ProtocolProcessor {
LOG.info("Found an existing connection with same client ID <{}>, forcing to close", msg.getClientID());
//clean the subscriptions if the old used a cleanSession = true
ServerChannel oldSession = m_clientIDs.get(msg.getClientID()).getSession();
boolean cleanSession = (Boolean) oldSession.getAttribute(NettyChannel.ATTR_KEY_CLEANSESSION);
if (cleanSession) {
//cleanup topic subscriptions
cleanSession(msg.getClientID());
}
ClientSession oldClientSession = m_sessionsStore.sessionForClient(msg.getClientID());
oldClientSession.disconnect();
oldSession.setAttribute(NettyChannel.ATTR_KEY_SESSION_STOLEN, true);
oldSession.close(false);
LOG.debug("Existing connection with same client ID <{}>, forced to close", msg.getClientID());
@@ -210,15 +207,6 @@ public class ProtocolProcessor {
m_willStore.put(msg.getClientID(), will);
}
subscriptions.activate(msg.getClientID());
//handle clean session flag
if (msg.isCleanSession()) {
//remove all prev subscriptions
//cleanup topic subscriptions
cleanSession(msg.getClientID());
}
ConnAckMessage okResp = new ConnAckMessage();
okResp.setReturnCode(ConnAckMessage.CONNECTION_ACCEPTED);
@@ -232,7 +220,11 @@ public class ProtocolProcessor {
if (!isSessionAlreadyStored) {
LOG.info("Create persistent session for clientID <{}>", msg.getClientID());
clientSession = m_sessionsStore.createNewSession(msg.getClientID());
clientSession = m_sessionsStore.createNewSession(msg.getClientID(), msg.isCleanSession());
}
clientSession.activate();
if (msg.isCleanSession()) {
clientSession.cleanSession();
}
LOG.info("Connected client ID <{}> with clean session {}", msg.getClientID(), msg.isCleanSession());
if (!msg.isCleanSession()) {
@@ -274,15 +266,6 @@ public class ProtocolProcessor {
m_messagesStore.removeMessageInSession(clientID, messageID);
}
private void cleanSession(String clientID) {
LOG.info("cleaning old saved subscriptions for client <{}>", clientID);
//remove from log all subscriptions
subscriptions.removeForClient(clientID);
//remove also the messages stored of type QoS1/2
m_messagesStore.dropMessagesInSession(clientID);
}
public void processPublish(ServerChannel session, PublishMessage msg) {
LOG.trace("PUB --PUBLISH--> SRV executePublish invoked with {}", msg);
String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID);
@@ -371,18 +354,18 @@ public class ProtocolProcessor {
if (qos.byteValue() > sub.getRequestedQos().byteValue()) {
qos = sub.getRequestedQos();
}
ClientSession targetSession = m_sessionsStore.sessionForClient(sub.getClientId());
LOG.debug("Broker republishing to client <{}> topic <{}> qos <{}>, active {}",
sub.getClientId(), sub.getTopicFilter(), qos, sub.isActive());
sub.getClientId(), sub.getTopicFilter(), qos, targetSession.isActive());
ByteBuffer message = origMessage.duplicate();
if (qos == AbstractMessage.QOSType.MOST_ONE && sub.isActive()) {
if (qos == AbstractMessage.QOSType.MOST_ONE && targetSession.isActive()) {
//QoS 0
//forwardPublishQoS0(sub.getClientId(), topic, qos, message, false);
sendPublish(sub.getClientId(), topic, qos, message, false, null);
} else {
//QoS 1 or 2
//if the target subscription is not clean session and is not connected => store it
if (!sub.isCleanSession() && !sub.isActive()) {
if (!sub.isCleanSession() && !targetSession.isActive()) {
//clone the event with matching clientID
PublishEvent newPublishEvt = new PublishEvent(topic, qos, message, retain, sub.getClientId(), messageID != null ? messageID : 0);
m_messagesStore.storePublishForFuture(newPublishEvt);
@@ -394,7 +377,7 @@ public class ProtocolProcessor {
m_messagesStore.storeTemporaryPublish(newPublishEvt, sub.getClientId(), messageID);
}
//publish
if (sub.isActive()) {
if (targetSession.isActive()) {
int messageId = m_messagesStore.nextPacketID(sub.getClientId());
sendPublish(sub.getClientId(), topic, qos, message, false, messageId);
}
@@ -510,8 +493,6 @@ public class ProtocolProcessor {
pubRelMessage.setMessageID(messageID);
pubRelMessage.setQos(AbstractMessage.QOSType.LEAST_ONE);
// m_clientIDs.get(clientID).getSession().write(pubRelMessage);
//disruptorPublish(new OutputMessagingEvent(m_clientIDs.get(clientID).getSession(), pubRelMessage));
session.write(pubRelMessage);
}
@@ -526,20 +507,18 @@ public class ProtocolProcessor {
public void processDisconnect(ServerChannel session, DisconnectMessage msg) throws InterruptedException {
String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID);
boolean cleanSession = (Boolean) session.getAttribute(NettyChannel.ATTR_KEY_CLEANSESSION);
if (cleanSession) {
//cleanup topic subscriptions
cleanSession(clientID);
}
LOG.info("DISCONNECT client <{}> with clean session {}", clientID, cleanSession);
ClientSession clientSession = m_sessionsStore.sessionForClient(clientID);
clientSession.disconnect();
m_clientIDs.remove(clientID);
session.close(true);
//de-activate the subscriptions for this ClientID
subscriptions.deactivate(clientID);
//cleanup the will store
m_willStore.remove(clientID);
LOG.info("DISCONNECT client <{}> with clean session {}", clientID, cleanSession);
m_interceptor.notifyClientDisconnected(clientID);
LOG.info("DISCONNECT client <{}> finished", clientID, cleanSession);
}
public void processConnectionLost(LostConnectionEvent evt) {
@@ -547,7 +526,8 @@ public class ProtocolProcessor {
//If already removed a disconnect message was already processed for this clientID
if (evt.sessionStolen && m_clientIDs.remove(clientID) != null) {
//de-activate the subscriptions for this ClientID
subscriptions.deactivate(clientID);
ClientSession clientSession = m_sessionsStore.sessionForClient(clientID);
clientSession.deactivate();
LOG.info("Lost connection with client <{}>", clientID);
}
//publish the Will message (if any) for the clientID
broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/Subscription.java
-15
View File
@@ -48,13 +48,6 @@ public class Subscription implements Serializable {
this.active = orig.active;
}
/**
* Factory method for empty subscriptions
*/
public static final Subscription createEmptySubscription(String clientId, boolean cleanSession) {
return new Subscription(clientId, "", QOSType.MOST_ONE, cleanSession);
}
public String getClientId() {
return clientId;
}
@@ -71,14 +64,6 @@ public class Subscription implements Serializable {
return this.cleanSession;
}
public boolean isActive() {
return active;
}
public void setActive(boolean active) {
this.active = active;
}
@Override
public boolean equals(Object o) {
if (this == o) return true;
broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java
+3 -54
View File
@@ -110,14 +110,14 @@ public class SubscriptionsStore {
for (Subscription subscription : subscriptions) {
LOG.debug("Re-subscribing {} to topic {}", subscription.getClientId(), subscription.getTopicFilter());
addDirect(subscription);
add(subscription);
}
if (LOG.isDebugEnabled()) {
LOG.debug("Finished loading. Subscription tree after {}", dumpTree());
}
}
protected void addDirect(Subscription newSubscription) {
public void add(Subscription newSubscription) {
TreeNode oldRoot;
NodeCouple couple;
do {
@@ -164,18 +164,6 @@ public class SubscriptionsStore {
return new NodeCouple(newRoot, current);
}
// public void save(ClientSession clientSession) {
// for(Subscription subscription : clientSession.getSubscriptions()) {
// add(subscription);
// }
// }
public void add(Subscription newSubscription) {
// m_sessionsStore.addNewSubscription(newSubscription);
addDirect(newSubscription);
}
public void removeSubscription(String topic, String clientID) {
TreeNode oldRoot;
NodeCouple couple;
@@ -212,47 +200,8 @@ public class SubscriptionsStore {
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));
//persist the update
m_sessionsStore.wipeSubscriptions(clientID);
}
/**
* Visit the topics tree to deactivate matching subscriptions with clientID.
* It's a mutating structure operation so create a new subscription tree (partial or total).
*/
public void deactivate(String clientID) {
LOG.debug("Disactivating subscriptions for clientID <{}>", clientID);
TreeNode oldRoot;
TreeNode newRoot;
do {
oldRoot = subscriptions.get();
newRoot = oldRoot.deactivate(clientID);
//spin lock repeating till we can, swap root, if can't swap just re-do the operation
} while(!subscriptions.compareAndSet(oldRoot, newRoot));
//persist the update
Set<Subscription> subs = newRoot.findAllByClientID(clientID);
m_sessionsStore.updateSubscriptions(clientID, subs);
}
/**
* Visit the topics tree to activate matching subscriptions with clientID.
* It's a mutating structure operation so create a new subscription tree (partial or total).
*/
public void activate(String clientID) {
LOG.debug("Activating subscriptions for clientID <{}>", clientID);
TreeNode oldRoot;
TreeNode newRoot;
do {
oldRoot = subscriptions.get();
newRoot = oldRoot.activate(clientID);
//spin lock repeating till we can, swap root, if can't swap just re-do the operation
} while(!subscriptions.compareAndSet(oldRoot, newRoot));
//persist the update
Set<Subscription> subs = newRoot.findAllByClientID(clientID);
m_sessionsStore.updateSubscriptions(clientID, subs);
}
/**
* Given a topic string return the clients subscriptions that matches it.
@@ -282,7 +231,7 @@ public class SubscriptionsStore {
subsForClient.put(sub.getClientId(), sub);
}
}
return /*matchingSubs*/new ArrayList<>(subsForClient.values());
return new ArrayList<>(subsForClient.values());
}
public boolean contains(Subscription sub) {
broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/TreeNode.java
-40
View File
@@ -173,46 +173,6 @@ class TreeNode {
return newSubRoot;
}
/**
* Deactivate all topic subscriptions for the given clientID.
* */
TreeNode deactivate(String clientID) {
TreeNode newSubRoot = this.copy();
for (Subscription s : newSubRoot.m_subscriptions) {
if (s.clientId.equals(clientID)) {
s.setActive(false);
}
}
//go deep
List<TreeNode> newChildren = new ArrayList<>(newSubRoot.m_children.size());
for (TreeNode child : newSubRoot.m_children) {
newChildren.add(child.deactivate(clientID));
}
newSubRoot.m_children = newChildren;
return newSubRoot;
}
/**
* Activate all topic subscriptions for the given clientID.
* */
TreeNode activate(String clientID) {
TreeNode newSubRoot = this.copy();
for (Subscription s : newSubRoot.m_subscriptions) {
if (s.clientId.equals(clientID)) {
s.setActive(true);
}
}
//go deep
List<TreeNode> newChildren = new ArrayList<>(newSubRoot.m_children.size());
for (TreeNode child : newSubRoot.m_children) {
newChildren.add(child.activate(clientID));
}
newSubRoot.m_children = newChildren;
return newSubRoot;
}
/**
* @return the set of subscriptions for the given client.
* */
broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java
+52 -8
View File
@@ -35,6 +35,7 @@ import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
import java.io.Serializable;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.*;
@@ -44,6 +45,20 @@ import java.util.concurrent.*;
*/
public class MapDBPersistentStore implements IMessagesStore, ISessionsStore {
/**
* This is a DTO used to persist minimal status (clean session and activation status) of
* a session.
* */
public static class PersistentSession implements Serializable {
public final boolean cleanSession;
public final boolean active;
public PersistentSession(boolean cleanSession, boolean active) {
this.cleanSession = cleanSession;
this.active = active;
}
}
private static final Logger LOG = LoggerFactory.getLogger(MapDBPersistentStore.class);
private ConcurrentMap<String, StoredMessage> m_retainedStore;
@@ -57,6 +72,7 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore {
private ConcurrentMap<String, StoredPublishEvent> m_qos2Store;
//persistent Map of clientID, set of Subscriptions
private ConcurrentMap<String, Set<Subscription>> m_persistentSubscriptions;
private ConcurrentMap<String, PersistentSession> m_persistentSessions;
private DB m_db;
private String m_storePath;
@@ -96,6 +112,7 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore {
m_inFlightIds = m_db.getHashMap("inflightPacketIDs");
m_persistentSubscriptions = m_db.getHashMap("subscriptions");
m_qos2Store = m_db.getHashMap("qos2Store");
m_persistentSessions = m_db.getHashMap("sessions");
m_scheduler.scheduleWithFixedDelay(new Runnable() {
@Override
public void run() {
@@ -127,7 +144,6 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore {
LOG.debug("searchMatching scanning all retained messages, presents are {}", m_retainedStore.size());
List<StoredMessage> results = new ArrayList<>();
for (Map.Entry<String, StoredMessage> entry : m_retainedStore.entrySet()) {
StoredMessage storedMsg = entry.getValue();
if (condition.match(entry.getKey())) {
@@ -245,6 +261,9 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore {
clientSubscriptions.remove(toBeRemoved);
}
m_persistentSubscriptions.put(clientID, clientSubscriptions);
if (LOG.isTraceEnabled()) {
LOG.trace("persisted subscriptions {}", m_persistentSubscriptions);
}
}
@Override
@@ -252,7 +271,7 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore {
LOG.debug("addNewSubscription invoked with subscription {}", newSubscription);
final String clientID = newSubscription.getClientId();
if (!m_persistentSubscriptions.containsKey(clientID)) {
LOG.debug("clientID {} is a newcome, creating it's subscriptions set", clientID);
LOG.debug("subscriptions for clientID <{}> not present, empty subscriptions set", clientID);
m_persistentSubscriptions.put(clientID, new HashSet<Subscription>());
}
@@ -301,24 +320,49 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore {
}
@Override
public ClientSession createNewSession(String clientID) {
LOG.debug("createNewSession for client <{}>", clientID);
if (m_persistentSubscriptions.containsKey(clientID)) {
public ClientSession createNewSession(String clientID, boolean cleanSession) {
LOG.debug("createNewSession for client <{}> with clean flag <{}>", clientID, cleanSession);
if (m_persistentSessions.containsKey(clientID)) {
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);
m_persistentSubscriptions.put(clientID, new HashSet<Subscription>());
return new ClientSession(clientID, this, this);
m_persistentSessions.putIfAbsent(clientID, new PersistentSession(cleanSession, false));
return new ClientSession(clientID, this, this, cleanSession);
}
@Override
public ClientSession sessionForClient(String clientID) {
if (!m_persistentSubscriptions.containsKey(clientID)) {
if (!m_persistentSessions.containsKey(clientID)) {
return null;
}
return new ClientSession(clientID, this, this);
PersistentSession storedSession = m_persistentSessions.get(clientID);
ClientSession clientSession = new ClientSession(clientID, this, this, storedSession.cleanSession);
if (storedSession.active) {
clientSession.activate();
}
return clientSession;
}
@Override
public void activate(String clientID) {
activationHelper(clientID, true);
}
@Override
public void deactivate(String clientID) {
activationHelper(clientID, false);
}
private void activationHelper(String clientID, boolean activation) {
PersistentSession storedSession = m_persistentSessions.get(clientID);
if (storedSession == null) {
throw new IllegalStateException((activation ? "activating" : "deactivating") + " a session never stored/created, clientID <"+ clientID + ">", null);
}
PersistentSession newStoredSession = new PersistentSession(storedSession.cleanSession, activation);
m_persistentSessions.put(clientID, newStoredSession);
}
@Override