From 19dac6e32b9ddf03c9b67f6e3d32ad726ca8f28d Mon Sep 17 00:00:00 2001 From: andsel Date: Sun, 8 Nov 2015 18:25:55 +0100 Subject: [PATCH] Refactored the disconenct and clean of session to use new ClientSession --- .../moquette/server/netty/NettyChannel.java | 2 +- .../eclipse/moquette/spi/ClientSession.java | 41 +++++++++++- .../eclipse/moquette/spi/ISessionsStore.java | 6 +- .../moquette/spi/impl/ProtocolProcessor.java | 62 +++++++------------ .../spi/impl/subscriptions/Subscription.java | 15 ----- .../subscriptions/SubscriptionsStore.java | 57 +---------------- .../spi/impl/subscriptions/TreeNode.java | 40 ------------ .../spi/persistence/MapDBPersistentStore.java | 60 +++++++++++++++--- .../moquette/server/IntegrationUtils.java | 20 +++++- .../server/ServerIntegrationFuseTest.java | 8 +-- .../server/ServerIntegrationPahoTest.java | 35 +++++------ .../ServerIntegrationQoSValidationTest.java | 8 +-- .../ServerIntegrationWebSocketTest.java | 5 +- ...erverLowlevelMessagesIntegrationTests.java | 5 +- .../server/ServerRestartIntegrationTest.java | 21 ++----- .../eclipse/moquette/server/TestCallback.java | 6 +- .../spi/impl/MemoryStorageService.java | 16 ++++- .../spi/impl/ProtocolProcessorTest.java | 31 ++++++---- .../subscriptions/SubscriptionsStoreTest.java | 9 +-- .../persistence/MapDBPersistentStoreTest.java | 16 ++--- .../MemorySessionStore.java | 40 ++++++++++-- 21 files changed, 239 insertions(+), 264 deletions(-) rename broker/src/test/java/org/eclipse/moquette/spi/{impl => persistence}/MemorySessionStore.java (71%) diff --git a/broker/src/main/java/org/eclipse/moquette/server/netty/NettyChannel.java b/broker/src/main/java/org/eclipse/moquette/server/netty/NettyChannel.java index de74c8da..3fb0342f 100644 --- a/broker/src/main/java/org/eclipse/moquette/server/netty/NettyChannel.java +++ b/broker/src/main/java/org/eclipse/moquette/server/netty/NettyChannel.java @@ -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 +"]"; } } diff --git a/broker/src/main/java/org/eclipse/moquette/spi/ClientSession.java b/broker/src/main/java/org/eclipse/moquette/spi/ClientSession.java index e07c33d2..5c8f277a 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/ClientSession.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/ClientSession.java @@ -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 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; + } } \ No newline at end of file diff --git a/broker/src/main/java/org/eclipse/moquette/spi/ISessionsStore.java b/broker/src/main/java/org/eclipse/moquette/spi/ISessionsStore.java index de9e6cc9..0369ed70 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/ISessionsStore.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/ISessionsStore.java @@ -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); } diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java index c3b41c80..19f9dc2c 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java @@ -93,7 +93,7 @@ public class ProtocolProcessor { private static final Logger LOG = LoggerFactory.getLogger(ProtocolProcessor.class); - private Map m_clientIDs = new ConcurrentHashMap<>(); + private Map 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 diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/Subscription.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/Subscription.java index 5c652f1a..ffd745fd 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/Subscription.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/Subscription.java @@ -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; diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java index 17b3fff6..256e970c 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java @@ -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 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 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) { diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/TreeNode.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/TreeNode.java index 2ca8da24..9c686733 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/TreeNode.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/TreeNode.java @@ -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 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 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. * */ diff --git a/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java b/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java index c0d9b113..beac521e 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java @@ -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 m_retainedStore; @@ -57,6 +72,7 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { private ConcurrentMap m_qos2Store; //persistent Map of clientID, set of Subscriptions private ConcurrentMap> m_persistentSubscriptions; + private ConcurrentMap 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 results = new ArrayList<>(); - for (Map.Entry 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()); } @@ -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()); - 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 diff --git a/broker/src/test/java/org/eclipse/moquette/server/IntegrationUtils.java b/broker/src/test/java/org/eclipse/moquette/server/IntegrationUtils.java index 05cca005..7a291aa6 100644 --- a/broker/src/test/java/org/eclipse/moquette/server/IntegrationUtils.java +++ b/broker/src/test/java/org/eclipse/moquette/server/IntegrationUtils.java @@ -15,18 +15,21 @@ */ package org.eclipse.moquette.server; +import org.eclipse.moquette.server.config.IConfig; + import java.io.File; import java.util.Properties; import static org.eclipse.moquette.commons.Constants.DEFAULT_MOQUETTE_STORE_MAP_DB_FILENAME; import static org.eclipse.moquette.commons.Constants.PERSISTENT_STORE_PROPERTY_NAME; +import static org.junit.Assert.assertFalse; /** * Used to carry integration configurations. * * Created by andrea on 4/7/15. */ -class IntegrationUtils { +public class IntegrationUtils { static String localMapDBPath() { String currentDir = System.getProperty("user.dir"); return currentDir + File.separator + DEFAULT_MOQUETTE_STORE_MAP_DB_FILENAME; @@ -37,4 +40,19 @@ class IntegrationUtils { testProperties.put(PERSISTENT_STORE_PROPERTY_NAME, IntegrationUtils.localMapDBPath()); return testProperties; } + + public static void cleanPersistenceFile(IConfig config) { + String fileName = config.getProperty(PERSISTENT_STORE_PROPERTY_NAME); + cleanPersistenceFile(fileName); + } + + public static void cleanPersistenceFile(String fileName) { + File dbFile = new File(fileName); + if (dbFile.exists()) { + dbFile.delete(); + new File(fileName + ".p").delete(); + new File(fileName + ".t").delete(); + } + assertFalse(dbFile.exists()); + } } diff --git a/broker/src/test/java/org/eclipse/moquette/server/ServerIntegrationFuseTest.java b/broker/src/test/java/org/eclipse/moquette/server/ServerIntegrationFuseTest.java index 0193491d..7b179b9e 100644 --- a/broker/src/test/java/org/eclipse/moquette/server/ServerIntegrationFuseTest.java +++ b/broker/src/test/java/org/eclipse/moquette/server/ServerIntegrationFuseTest.java @@ -62,9 +62,6 @@ public class ServerIntegrationFuseTest { @After public void tearDown() throws Exception { -// if (m_mqtt.isConnected()) { -// m_mqtt.disconnect(); -// } if (m_subscriber != null) { m_subscriber.disconnect(); } @@ -74,10 +71,7 @@ public class ServerIntegrationFuseTest { } m_server.stopServer(); - File dbFile = new File(m_config.getProperty(PERSISTENT_STORE_PROPERTY_NAME)); - if (dbFile.exists()) { - dbFile.delete(); - } + IntegrationUtils.cleanPersistenceFile(m_config); } @Test diff --git a/broker/src/test/java/org/eclipse/moquette/server/ServerIntegrationPahoTest.java b/broker/src/test/java/org/eclipse/moquette/server/ServerIntegrationPahoTest.java index 7cf776d1..fe70981c 100644 --- a/broker/src/test/java/org/eclipse/moquette/server/ServerIntegrationPahoTest.java +++ b/broker/src/test/java/org/eclipse/moquette/server/ServerIntegrationPahoTest.java @@ -30,7 +30,6 @@ import java.io.File; import java.io.IOException; import java.util.Properties; -import static org.eclipse.moquette.commons.Constants.*; import static org.junit.Assert.*; public class ServerIntegrationPahoTest { @@ -83,11 +82,7 @@ public class ServerIntegrationPahoTest { private void stopServer() { m_server.stopServer(); - File dbFile = new File(m_config.getProperty(PERSISTENT_STORE_PROPERTY_NAME)); - if (dbFile.exists()) { - dbFile.delete(); - } - assertFalse(dbFile.exists()); + IntegrationUtils.cleanPersistenceFile(m_config); } @Test @@ -159,7 +154,7 @@ public class ServerIntegrationPahoTest { m_client.connect(); m_client.publish("/topic", "Test my payload".getBytes(), 0, false); - assertNull(m_callback.getMessage(false)); + assertEquals("Test my payload", new String(m_callback.getMessage(false).getPayload())); } @Test @@ -265,7 +260,7 @@ public class ServerIntegrationPahoTest { assertEquals("Hello MQTT", m_callback.getMessage(true).toString()); } - + @Test public void checkReceivePublishedMessage_after_a_reconnect_with_notCleanSession() throws Exception { LOG.info("*** checkReceivePublishedMessage_after_a_reconnect_with_notCleanSession ***"); @@ -274,19 +269,19 @@ public class ServerIntegrationPahoTest { m_client.connect(options); m_client.subscribe("/topic", 1); m_client.disconnect(); - + m_client.connect(options); m_client.subscribe("/topic", 1); - + //publish a QoS 1 message another client publish a message on the topic - publishFromAnotherClient("/topic", "Hello MQTT".getBytes(), 1); - + publishFromAnotherClient("/topic", "Hello MQTT".getBytes(), 1); + //Verify that after a reconnection the client receive the message MqttMessage message = m_callback.getMessage(true); assertNotNull(message); assertEquals("Hello MQTT", message.toString()); } - + private void publishFromAnotherClient(String topic, byte[] payload, int qos) throws Exception { IMqttClient anotherClient = new MqttClient("tcp://localhost:1883", "TestClientPUB", s_pubDataStore); anotherClient.connect(); @@ -307,7 +302,7 @@ public class ServerIntegrationPahoTest { publishFromAnotherClient("/topic", "Hello MQTT".getBytes(), 2); m_callback.reinit(); m_client.connect(options); - + MqttMessage message = m_callback.getMessage(true); assertEquals("Hello MQTT", message.toString()); assertEquals(2, message.getQos()); @@ -326,13 +321,13 @@ public class ServerIntegrationPahoTest { publishFromAnotherClient("/topic", "Hello MQTT".getBytes(), 2); m_callback.reinit(); m_client.connect(options); - + assertNotNull(m_callback); MqttMessage message = m_callback.getMessage(true); assertNotNull(message); assertEquals("Hello MQTT", message.toString()); } - + @Test public void avoidMultipleNotificationsAfterMultipleReconnection_cleanSessionFalseQoS1() throws Exception { LOG.info("*** avoidMultipleNotificationsAfterMultipleReconnection_cleanSessionFalseQoS1, issue #16 ***"); @@ -341,20 +336,20 @@ public class ServerIntegrationPahoTest { m_client.connect(options); m_client.subscribe("/topic", 1); m_client.disconnect(); - + publishFromAnotherClient("/topic", "Hello MQTT 1".getBytes(), 1); m_callback.reinit(); m_client.connect(options); - + assertNotNull(m_callback); MqttMessage message = m_callback.getMessage(true); assertNotNull(message); assertEquals("Hello MQTT 1", message.toString()); m_client.disconnect(); - + //publish other message publishFromAnotherClient("/topic", "Hello MQTT 2".getBytes(), 1); - + //reconnect the second time m_callback.reinit(); m_client.connect(options); diff --git a/broker/src/test/java/org/eclipse/moquette/server/ServerIntegrationQoSValidationTest.java b/broker/src/test/java/org/eclipse/moquette/server/ServerIntegrationQoSValidationTest.java index 48044a29..2920151b 100644 --- a/broker/src/test/java/org/eclipse/moquette/server/ServerIntegrationQoSValidationTest.java +++ b/broker/src/test/java/org/eclipse/moquette/server/ServerIntegrationQoSValidationTest.java @@ -34,7 +34,6 @@ import java.io.File; import java.io.IOException; import java.util.Properties; -import static org.eclipse.moquette.commons.Constants.PERSISTENT_STORE_PROPERTY_NAME; import static org.junit.Assert.assertEquals; /** @@ -78,8 +77,6 @@ public class ServerIntegrationQoSValidationTest { m_subscriber.connect(); m_publisher = new MqttClient("tcp://localhost:1883", "Publisher", s_pubDataStore); -// m_callback = new TestCallback(); -// m_subscriber.setCallback(m_callback); m_publisher.connect(); } @@ -94,10 +91,7 @@ public class ServerIntegrationQoSValidationTest { } m_server.stopServer(); - File dbFile = new File(m_config.getProperty(PERSISTENT_STORE_PROPERTY_NAME)); - if (dbFile.exists()) { - dbFile.delete(); - } + IntegrationUtils.cleanPersistenceFile(m_config); } @Test diff --git a/broker/src/test/java/org/eclipse/moquette/server/ServerIntegrationWebSocketTest.java b/broker/src/test/java/org/eclipse/moquette/server/ServerIntegrationWebSocketTest.java index a04420f6..c740d8db 100644 --- a/broker/src/test/java/org/eclipse/moquette/server/ServerIntegrationWebSocketTest.java +++ b/broker/src/test/java/org/eclipse/moquette/server/ServerIntegrationWebSocketTest.java @@ -66,10 +66,7 @@ public class ServerIntegrationWebSocketTest { client.stop(); m_server.stopServer(); - File dbFile = new File(m_config.getProperty(PERSISTENT_STORE_PROPERTY_NAME)); - if (dbFile.exists()) { - dbFile.delete(); - } + IntegrationUtils.cleanPersistenceFile(m_config); } @Test diff --git a/broker/src/test/java/org/eclipse/moquette/server/ServerLowlevelMessagesIntegrationTests.java b/broker/src/test/java/org/eclipse/moquette/server/ServerLowlevelMessagesIntegrationTests.java index e79cd1d5..8619dc5d 100644 --- a/broker/src/test/java/org/eclipse/moquette/server/ServerLowlevelMessagesIntegrationTests.java +++ b/broker/src/test/java/org/eclipse/moquette/server/ServerLowlevelMessagesIntegrationTests.java @@ -73,10 +73,7 @@ public class ServerLowlevelMessagesIntegrationTests { Thread.sleep(300); //to let the close event pass before server stop event m_server.stopServer(); LOG.debug("After asked server to stop"); - File dbFile = new File(m_config.getProperty(PERSISTENT_STORE_PROPERTY_NAME)); - if (dbFile.exists()) { - dbFile.delete(); - } + IntegrationUtils.cleanPersistenceFile(m_config); } @Test diff --git a/broker/src/test/java/org/eclipse/moquette/server/ServerRestartIntegrationTest.java b/broker/src/test/java/org/eclipse/moquette/server/ServerRestartIntegrationTest.java index c309bc3e..4bc16a6e 100644 --- a/broker/src/test/java/org/eclipse/moquette/server/ServerRestartIntegrationTest.java +++ b/broker/src/test/java/org/eclipse/moquette/server/ServerRestartIntegrationTest.java @@ -21,15 +21,9 @@ import org.fusesource.mqtt.client.*; import org.junit.After; import org.junit.Before; import org.junit.Test; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.io.File; import java.io.IOException; import java.util.Properties; import java.util.concurrent.TimeUnit; - -import static org.eclipse.moquette.commons.Constants.PERSISTENT_STORE_PROPERTY_NAME; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNull; @@ -38,8 +32,7 @@ import static org.junit.Assert.assertNull; * @author andrea */ public class ServerRestartIntegrationTest { - private static Logger LOG = LoggerFactory.getLogger(ServerRestartIntegrationTest.class); - + Server m_server; MQTT m_mqtt; BlockingConnection m_subscriber; @@ -74,10 +67,7 @@ public class ServerRestartIntegrationTest { } m_server.stopServer(); - File dbFile = new File(m_config.getProperty(PERSISTENT_STORE_PROPERTY_NAME)); - if (dbFile.exists()) { - dbFile.delete(); - } + IntegrationUtils.cleanPersistenceFile(m_config); } @@ -93,11 +83,8 @@ public class ServerRestartIntegrationTest { //shutdown the server m_server.stopServer(); - File dbFile = new File(m_config.getProperty(PERSISTENT_STORE_PROPERTY_NAME)); - if (dbFile.exists()) { - dbFile.delete(); - } - + IntegrationUtils.cleanPersistenceFile(m_config); + //restart the server m_server.startServer(IntegrationUtils.prepareTestPropeties()); diff --git a/broker/src/test/java/org/eclipse/moquette/server/TestCallback.java b/broker/src/test/java/org/eclipse/moquette/server/TestCallback.java index c38c9420..fa3d5cf3 100644 --- a/broker/src/test/java/org/eclipse/moquette/server/TestCallback.java +++ b/broker/src/test/java/org/eclipse/moquette/server/TestCallback.java @@ -38,7 +38,7 @@ class TestCallback implements MqttCallback { throw new IllegalStateException("Elapsed the timeout to get the result"); } } catch (InterruptedException e) { - e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates. + e.printStackTrace(); } return m_message; } @@ -47,7 +47,7 @@ class TestCallback implements MqttCallback { try { m_latch.await(5, TimeUnit.SECONDS); } catch (InterruptedException e) { - e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates. + e.printStackTrace(); } return m_topic; } @@ -59,7 +59,6 @@ class TestCallback implements MqttCallback { } public void connectionLost(Throwable cause) { - //To change body of implemented methods use File | Settings | File Templates. } public void messageArrived(String topic, MqttMessage message) throws Exception { @@ -69,6 +68,5 @@ class TestCallback implements MqttCallback { } public void deliveryComplete(IMqttDeliveryToken token) { - //To change body of implemented methods use File | Settings | File Templates. } } diff --git a/broker/src/test/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java b/broker/src/test/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java index 60dfc9ac..bca02ecb 100644 --- a/broker/src/test/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java +++ b/broker/src/test/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java @@ -18,11 +18,11 @@ package org.eclipse.moquette.spi.impl; import java.nio.ByteBuffer; import java.util.*; -import jdk.nashorn.internal.objects.NativeJava; import org.eclipse.moquette.spi.ClientSession; import org.eclipse.moquette.spi.IMatchingCondition; import org.eclipse.moquette.spi.IMessagesStore; import org.eclipse.moquette.spi.impl.events.PublishEvent; +import org.eclipse.moquette.spi.persistence.MemorySessionStore; import org.eclipse.moquette.spi.impl.subscriptions.Subscription; import org.eclipse.moquette.proto.messages.AbstractMessage; @@ -211,12 +211,22 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore { } @Override - public ClientSession createNewSession(String clientID) { - return m_sessionsStore.createNewSession(clientID); + public ClientSession createNewSession(String clientID, boolean cleanSession) { + return m_sessionsStore.createNewSession(clientID, cleanSession); } @Override public ClientSession sessionForClient(String clientID) { return m_sessionsStore.sessionForClient(clientID); } + + @Override + public void activate(String clientID) { + m_sessionsStore.activate(clientID); + } + + @Override + public void deactivate(String clientID) { + m_sessionsStore.deactivate(clientID); + } } diff --git a/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java index c08f80d4..970d7cdc 100644 --- a/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java @@ -27,6 +27,7 @@ import org.eclipse.moquette.spi.IMatchingCondition; import org.eclipse.moquette.spi.IMessagesStore; import org.eclipse.moquette.spi.ISessionsStore; import org.eclipse.moquette.spi.impl.events.PublishEvent; +import org.eclipse.moquette.spi.persistence.MemorySessionStore; import org.eclipse.moquette.spi.impl.security.PermitAllAuthorizator; import org.eclipse.moquette.spi.impl.subscriptions.Subscription; import org.eclipse.moquette.spi.impl.subscriptions.SubscriptionsStore; @@ -117,6 +118,7 @@ public class ProtocolProcessorTest { ConnectMessage connectMessage = new ConnectMessage(); connectMessage.setProtocolVersion((byte) 3); connectMessage.setClientID(FAKE_CLIENT_ID); + m_sessionStore.createNewSession(FAKE_CLIENT_ID, false); connectMessage.setCleanSession(subscription.isCleanSession()); m_processor.processConnect(m_session, connectMessage); @@ -207,7 +209,7 @@ public class ProtocolProcessorTest { msg.addSubscription(new SubscribeMessage.Couple(AbstractMessage.QOSType.MOST_ONE.byteValue(), FAKE_TOPIC)); m_session.setAttribute(NettyChannel.ATTR_KEY_CLIENTID, FAKE_CLIENT_ID); m_session.setAttribute(NettyChannel.ATTR_KEY_CLEANSESSION, false); - m_sessionStore.createNewSession(FAKE_CLIENT_ID); + m_sessionStore.createNewSession(FAKE_CLIENT_ID, false); m_processor.processSubscribe(m_session, msg/*, FAKE_CLIENT_ID, false*/); //Verify @@ -222,7 +224,7 @@ public class ProtocolProcessorTest { msg.addSubscription(new SubscribeMessage.Couple(AbstractMessage.QOSType.MOST_ONE.byteValue(), FAKE_TOPIC)); m_session.setAttribute(NettyChannel.ATTR_KEY_CLIENTID, FAKE_CLIENT_ID); m_session.setAttribute(NettyChannel.ATTR_KEY_CLEANSESSION, false); - m_sessionStore.createNewSession(FAKE_CLIENT_ID); + m_sessionStore.createNewSession(FAKE_CLIENT_ID, false); assertEquals(0, subscriptions.size()); m_processor.processSubscribe(m_session, msg); @@ -243,7 +245,7 @@ public class ProtocolProcessorTest { msg.addSubscription(new SubscribeMessage.Couple(AbstractMessage.QOSType.MOST_ONE.byteValue(), BAD_FORMATTED_TOPIC)); m_session.setAttribute(NettyChannel.ATTR_KEY_CLIENTID, FAKE_CLIENT_ID); m_session.setAttribute(NettyChannel.ATTR_KEY_CLEANSESSION, false); - m_sessionStore.createNewSession(FAKE_CLIENT_ID); + m_sessionStore.createNewSession(FAKE_CLIENT_ID, false); assertEquals(0, subscriptions.size()); //Exercise @@ -279,7 +281,7 @@ public class ProtocolProcessorTest { @Test public void testPublishOfRetainedMessage_afterNewSubscription() throws Exception { final CountDownLatch publishRecvSignal = new CountDownLatch(1); - /*ServerChannel */m_session = new DummyChannel() { + m_session = new DummyChannel() { @Override public void write(Object value) { try { @@ -370,10 +372,12 @@ public class ProtocolProcessorTest { } @Test - public void publishNoPublishToInactiveSubscriptions() { + public void publishNoPublishToInactiveSession() { + //create an inactive session for Subscriber + m_sessionStore.createNewSession("Subscriber", false).deactivate(); + SubscriptionsStore mockedSubscriptions = mock(SubscriptionsStore.class); Subscription inactiveSub = new Subscription("Subscriber", "/topic", QOSType.LEAST_ONE, false); - inactiveSub.setActive(false); List inactiveSubscriptions = Arrays.asList(inactiveSub); when(mockedSubscriptions.matches(eq("/topic"))).thenReturn(inactiveSubscriptions); m_processor = new ProtocolProcessor(); @@ -396,14 +400,16 @@ public class ProtocolProcessorTest { @Test public void publishToAnInactiveSubscriptionsCleanSession() { + //create an inactive session for Subscriber + m_sessionStore.createNewSession("Subscriber", false).deactivate(); SubscriptionsStore mockedSubscriptions = mock(SubscriptionsStore.class); Subscription inactiveSub = new Subscription("Subscriber", "/topic", QOSType.LEAST_ONE, true); - inactiveSub.setActive(false); List inactiveSubscriptions = Arrays.asList(inactiveSub); when(mockedSubscriptions.matches(eq("/topic"))).thenReturn(inactiveSubscriptions); m_processor = new ProtocolProcessor(); - m_processor.init(mockedSubscriptions, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); - + m_processor.init(mockedSubscriptions, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator(), + NO_OBSERVERS_INTERCEPTOR); + //Exercise ByteBuffer buffer = ByteBuffer.allocate(5).put("Hello".getBytes()); PublishMessage msg = new PublishMessage(); @@ -463,11 +469,14 @@ public class ProtocolProcessorTest { ByteBuffer payload = ByteBuffer.wrap("Hello world MQTT!!".getBytes()); PublishEvent forwardPublish = new PublishEvent("a/b", QOSType.EXACTLY_ONCE, payload, true, "Publisher", 1); IMessagesStore memoryMessageStore = new MemoryStorageService(); + ISessionsStore sessionsStore = new MemorySessionStore(memoryMessageStore); + sessionsStore.createNewSession("Sub A", false).activate(); + sessionsStore.createNewSession("Sub B", false).activate(); Subscription subQos1 = new Subscription("Sub A", "a/b", QOSType.LEAST_ONE, false); Subscription subQos2 = new Subscription("Sub B", "a/+", QOSType.EXACTLY_ONCE, false); SubscriptionsStore subscriptions = new SubscriptionsStore(); - subscriptions.init(new MemorySessionStore(memoryMessageStore)); + subscriptions.init(sessionsStore); subscriptions.add(subQos1); subscriptions.add(subQos2); @@ -479,7 +488,7 @@ public class ProtocolProcessorTest { publishedForwarded.add(new PublishEvent(topic, qos, message, retained, clientId, messageID)); } }; - processor.init(subscriptions, memoryMessageStore, null, null, true, null, NO_OBSERVERS_INTERCEPTOR); + processor.init(subscriptions, memoryMessageStore, sessionsStore, null, true, null, NO_OBSERVERS_INTERCEPTOR); //Exercise processor.forward2Subscribers(forwardPublish); diff --git a/broker/src/test/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java b/broker/src/test/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java index e7854754..07523534 100644 --- a/broker/src/test/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java +++ b/broker/src/test/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java @@ -58,11 +58,6 @@ public class SubscriptionsStoreTest { assertEqualsSeq(asArray(Token.EMPTY, Token.EMPTY), tokens); } -// @Test(expected = ParseException.class) -// public void testSplitTopicTwinsSlashAvoided() throws ParseException { -// store.parseTopic("/finance//stock/ibm"); -// } - @Test public void testParseTopicMultiValid() throws ParseException { List tokens = store.parseTopic("finance/stock/#"); @@ -355,8 +350,8 @@ public class SubscriptionsStoreTest { //subscribe a not active clientID1 to /topic Subscription slashSub = new Subscription("FAKE_CLI_ID_1", "/topic", AbstractMessage.QOSType.MOST_ONE, false); aStore.add(slashSub); - aStore.deactivate(slashSub.getClientId()); - + memStore.createNewSession("FAKE_CLI_ID_1", true).deactivate(); + //subscribe an active clientID2 to /topic Subscription slashSub2 = new Subscription("FAKE_CLI_ID_2", "/topic", AbstractMessage.QOSType.MOST_ONE, false); aStore.add(slashSub2); diff --git a/broker/src/test/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStoreTest.java b/broker/src/test/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStoreTest.java index 4977854a..5d970ba0 100644 --- a/broker/src/test/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStoreTest.java +++ b/broker/src/test/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStoreTest.java @@ -16,8 +16,8 @@ package org.eclipse.moquette.spi.persistence; import org.eclipse.moquette.proto.messages.AbstractMessage; +import org.eclipse.moquette.server.IntegrationUtils; import org.eclipse.moquette.spi.impl.subscriptions.Subscription; -import org.eclipse.moquette.spi.persistence.MapDBPersistentStore; import org.junit.After; import org.junit.Before; import org.junit.Test; @@ -36,12 +36,10 @@ import static org.junit.Assert.*; public class MapDBPersistentStoreTest { MapDBPersistentStore m_storageService; - + @Before public void setUp() throws Exception { - File dbFile = new File(DEFAULT_PERSISTENT_PATH); - assertFalse(String.format("The DB storage file %s already exists", DEFAULT_PERSISTENT_PATH), dbFile.exists()); - + IntegrationUtils.cleanPersistenceFile(DEFAULT_PERSISTENT_PATH); m_storageService = new MapDBPersistentStore(DEFAULT_PERSISTENT_PATH); m_storageService.initStore(); } @@ -51,12 +49,8 @@ public class MapDBPersistentStoreTest { if (m_storageService != null) { m_storageService.close(); } - - File dbFile = new File(DEFAULT_PERSISTENT_PATH); - if (dbFile.exists()) { - assertTrue("Error deleting the moquette db file " + DEFAULT_PERSISTENT_PATH, dbFile.delete()); - } - assertFalse(dbFile.exists()); + + IntegrationUtils.cleanPersistenceFile(DEFAULT_PERSISTENT_PATH); } @Test diff --git a/broker/src/test/java/org/eclipse/moquette/spi/impl/MemorySessionStore.java b/broker/src/test/java/org/eclipse/moquette/spi/persistence/MemorySessionStore.java similarity index 71% rename from broker/src/test/java/org/eclipse/moquette/spi/impl/MemorySessionStore.java rename to broker/src/test/java/org/eclipse/moquette/spi/persistence/MemorySessionStore.java index f9df08e8..cd8db2f6 100644 --- a/broker/src/test/java/org/eclipse/moquette/spi/impl/MemorySessionStore.java +++ b/broker/src/test/java/org/eclipse/moquette/spi/persistence/MemorySessionStore.java @@ -13,12 +13,13 @@ * * You may elect to redistribute this code under either of these licenses. */ -package org.eclipse.moquette.spi.impl; +package org.eclipse.moquette.spi.persistence; import org.eclipse.moquette.spi.ClientSession; import org.eclipse.moquette.spi.IMessagesStore; import org.eclipse.moquette.spi.ISessionsStore; import org.eclipse.moquette.spi.impl.subscriptions.Subscription; +import static org.eclipse.moquette.spi.persistence.MapDBPersistentStore.PersistentSession; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -33,6 +34,8 @@ public class MemorySessionStore implements ISessionsStore { private Map> m_persistentSubscriptions = new HashMap<>(); + private Map m_persistentSessions = new HashMap<>(); + private final IMessagesStore m_messagesStore; public MemorySessionStore(IMessagesStore messagesStore) { @@ -90,24 +93,30 @@ public class MemorySessionStore implements ISessionsStore { } @Override - public ClientSession createNewSession(String clientID) { + public ClientSession createNewSession(String clientID, boolean cleanSession) { LOG.debug("createNewSession for client <{}>", clientID); - if (m_persistentSubscriptions.containsKey(clientID)) { + 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()); - return new ClientSession(clientID, m_messagesStore, this); + m_persistentSessions.put(clientID, new PersistentSession(cleanSession, false)); + return new ClientSession(clientID, m_messagesStore, this, cleanSession); } @Override public ClientSession sessionForClient(String clientID) { - if (!m_persistentSubscriptions.containsKey(clientID)) { + if (!m_persistentSessions.containsKey(clientID)) { return null; } - return new ClientSession(clientID, m_messagesStore, this); + PersistentSession storedSession = m_persistentSessions.get(clientID); + ClientSession clientSession = new ClientSession(clientID, m_messagesStore, this, storedSession.cleanSession); + if (storedSession.active) { + clientSession.activate(); + } + return clientSession; } @Override @@ -118,4 +127,23 @@ public class MemorySessionStore implements ISessionsStore { } return allSubscriptions; } + + @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); + } }