diff --git a/ChangeLog.txt b/ChangeLog.txt index 1be85c94..57bc958b 100644 --- a/ChangeLog.txt +++ b/ChangeLog.txt @@ -5,6 +5,8 @@ Version 0.6: Integrated SSL support (issue #41) Fixed issue #40, check Logging inside the Felix OSGi Fixed issue #46, implement correct behaviour for Will stuff + Splitted stores to persist messages and sessions into different interfaces + Renamed some methods in sessions store Version 0.5: Fix issue 20 (no flush of message on PINGREQ) diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessagesStore.java similarity index 91% rename from broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java rename to broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessagesStore.java index bca381dc..d3fafe5a 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessagesStore.java @@ -18,15 +18,15 @@ package org.dna.mqtt.moquette.messaging.spi; import java.nio.ByteBuffer; import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent; import org.dna.mqtt.moquette.proto.messages.AbstractMessage; -import static org.dna.mqtt.moquette.messaging.spi.impl.HawtDBStorageService.StoredMessage; +import static org.dna.mqtt.moquette.messaging.spi.impl.HawtDBPersistentStore.StoredMessage; import java.util.Collection; import java.util.List; /** - * Defines the SPI to be implemented by a StorageService that handle persistence of messages and subscriptions. + * Defines the SPI to be implemented by a StorageService that handle persistence of messages */ -public interface IStorageService extends IPersistentSubscriptionStore { +public interface IMessagesStore { /** * Used to initialize all persistent store structures diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IPersistentSubscriptionStore.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/ISessionsStore.java similarity index 95% rename from broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IPersistentSubscriptionStore.java rename to broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/ISessionsStore.java index e81cfd83..ad4add5b 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IPersistentSubscriptionStore.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/ISessionsStore.java @@ -23,7 +23,7 @@ import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription; * * @author andrea */ -public interface IPersistentSubscriptionStore { +public interface ISessionsStore { /** * Add a new subscription to the session diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBPersistentStore.java similarity index 97% rename from broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java rename to broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBPersistentStore.java index c231b5a1..905c9666 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBPersistentStore.java @@ -22,7 +22,8 @@ import java.nio.ByteBuffer; import java.util.*; import org.dna.mqtt.moquette.MQTTException; import org.dna.mqtt.moquette.messaging.spi.IMatchingCondition; -import org.dna.mqtt.moquette.messaging.spi.IStorageService; +import org.dna.mqtt.moquette.messaging.spi.ISessionsStore; +import org.dna.mqtt.moquette.messaging.spi.IMessagesStore; import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent; import org.dna.mqtt.moquette.messaging.spi.impl.storage.StoredPublishEvent; import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription; @@ -35,9 +36,9 @@ import org.slf4j.LoggerFactory; /** - * Implementation of IStorageService backed by HawtDB + * Implementation of IMessagesStore and ISessionsStore backed by HawtDB */ -public class HawtDBStorageService implements IStorageService { +public class HawtDBPersistentStore implements IMessagesStore, ISessionsStore { public static class StoredMessage implements Serializable { AbstractMessage.QOSType m_qos; @@ -64,7 +65,7 @@ public class HawtDBStorageService implements IStorageService { } - private static final Logger LOG = LoggerFactory.getLogger(HawtDBStorageService.class); + private static final Logger LOG = LoggerFactory.getLogger(HawtDBPersistentStore.class); private MultiIndexFactory m_multiIndexFactory; private PageFileFactory pageFactory; @@ -81,7 +82,7 @@ public class HawtDBStorageService implements IStorageService { //persistent Map of clientID, set of Subscriptions private SortedIndex> m_persistentSubscriptions; - public HawtDBStorageService() { + public HawtDBPersistentStore() { String storeFile = Server.STORAGE_FILE_PATH; pageFactory = new PageFileFactory(); diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java index 88daec81..cf6596b3 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java @@ -18,7 +18,7 @@ package org.dna.mqtt.moquette.messaging.spi.impl; import java.nio.ByteBuffer; import java.util.ArrayList; import org.dna.mqtt.moquette.messaging.spi.IMatchingCondition; -import org.dna.mqtt.moquette.messaging.spi.IStorageService; +import org.dna.mqtt.moquette.messaging.spi.IMessagesStore; import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent; import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription; import org.dna.mqtt.moquette.proto.messages.AbstractMessage; @@ -29,15 +29,16 @@ import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Set; +import org.dna.mqtt.moquette.messaging.spi.ISessionsStore; import org.slf4j.Logger; import org.slf4j.LoggerFactory; /** */ -public class MemoryStorageService implements IStorageService { +public class MemoryStorageService implements IMessagesStore, ISessionsStore { private Map> m_persistentSubscriptions = new HashMap>(); - private Map m_retainedStore = new HashMap(); + private Map m_retainedStore = new HashMap(); //TODO move in a multimap because only Qos1 and QoS2 are stored here and they have messageID(key of secondary map) private Map> m_persistentMessageStore = new HashMap>(); private Map m_inflightStore = new HashMap(); @@ -57,17 +58,17 @@ public class MemoryStorageService implements IStorageService { //store the message to the topic byte[] raw = new byte[message.remaining()]; message.get(raw); - m_retainedStore.put(topic, new HawtDBStorageService.StoredMessage(raw, qos, topic)); + m_retainedStore.put(topic, new HawtDBPersistentStore.StoredMessage(raw, qos, topic)); } } - public Collection searchMatching(IMatchingCondition condition) { + public Collection searchMatching(IMatchingCondition condition) { LOG.debug("searchMatching scanning all retained messages, presents are {}", m_retainedStore.size()); - List results = new ArrayList(); + List results = new ArrayList(); - for (Map.Entry entry : m_retainedStore.entrySet()) { - HawtDBStorageService.StoredMessage storedMsg = entry.getValue(); + for (Map.Entry entry : m_retainedStore.entrySet()) { + HawtDBPersistentStore.StoredMessage storedMsg = entry.getValue(); if (condition.match(entry.getKey())) { results.add(storedMsg); } diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java index 8b072410..05f5159c 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java @@ -27,7 +27,8 @@ import java.util.Map; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import org.dna.mqtt.moquette.messaging.spi.IMatchingCondition; -import org.dna.mqtt.moquette.messaging.spi.IStorageService; +import org.dna.mqtt.moquette.messaging.spi.ISessionsStore; +import org.dna.mqtt.moquette.messaging.spi.IMessagesStore; import org.dna.mqtt.moquette.messaging.spi.impl.events.MessagingEvent; import org.dna.mqtt.moquette.messaging.spi.impl.events.OutputMessagingEvent; import org.dna.mqtt.moquette.messaging.spi.impl.events.PubAckEvent; @@ -98,7 +99,8 @@ class ProtocolProcessor implements EventHandler { private Map m_clientIDs = new HashMap(); private SubscriptionsStore subscriptions; - private IStorageService m_storageService; + private IMessagesStore m_storageService; + private ISessionsStore m_sessionsStore; private IAuthenticator m_authenticator; //maps clientID to Will testament, if specified on CONNECT private Map m_willStore = new HashMap(); @@ -114,14 +116,17 @@ class ProtocolProcessor implements EventHandler { * clients subscriptions. * @param storageService the persistent store to use for save/load of messages * for QoS1 and QoS2 handling. + * @param sessionsStore the clients sessions store, used to persist subscriptions. * @param authenticator the authenticator used in connect messages */ - void init(SubscriptionsStore subscriptions, IStorageService storageService, + void init(SubscriptionsStore subscriptions, IMessagesStore storageService, + ISessionsStore sessionsStore, IAuthenticator authenticator) { //m_clientIDs = clientIDs; this.subscriptions = subscriptions; m_authenticator = authenticator; m_storageService = storageService; + m_sessionsStore = sessionsStore; //init the output ringbuffer m_executor = Executors.newFixedThreadPool(1); @@ -246,7 +251,7 @@ class ProtocolProcessor implements EventHandler { private void cleanSession(String clientID) { LOG.info("cleaning old saved subscriptions for client <{}>", clientID); //remove from log all subscriptions - m_storageService.wipeSubscriptions(clientID); + m_sessionsStore.wipeSubscriptions(clientID); subscriptions.removeForClient(clientID); //remove also the messages stored of type QoS1/2 @@ -537,17 +542,17 @@ class ProtocolProcessor implements EventHandler { newSubscription.getClientId(), topic, AbstractMessage.QOSType.formatQoS(newSubscription.getRequestedQos())); String clientID = newSubscription.getClientId(); - m_storageService.addNewSubscription(newSubscription, clientID); + m_sessionsStore.addNewSubscription(newSubscription, clientID); subscriptions.add(newSubscription); //scans retained messages to be published to the new subscription - Collection messages = m_storageService.searchMatching(new IMatchingCondition() { + Collection messages = m_storageService.searchMatching(new IMatchingCondition() { public boolean match(String key) { return SubscriptionsStore.matchTopics(key, topic); } }); - for (HawtDBStorageService.StoredMessage storedMsg : messages) { + for (HawtDBPersistentStore.StoredMessage storedMsg : messages) { //fire the as retained the message LOG.debug("send publish message for topic {}", topic); sendPublish(newSubscription.getClientId(), storedMsg.getTopic(), storedMsg.getQos(), storedMsg.getPayload(), true); diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java index 73ea09d7..26443a26 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java @@ -26,7 +26,8 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import org.dna.mqtt.moquette.messaging.spi.IMessaging; -import org.dna.mqtt.moquette.messaging.spi.IStorageService; +import org.dna.mqtt.moquette.messaging.spi.ISessionsStore; +import org.dna.mqtt.moquette.messaging.spi.IMessagesStore; import org.dna.mqtt.moquette.messaging.spi.impl.events.*; import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription; import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.SubscriptionsStore; @@ -56,7 +57,8 @@ public class SimpleMessaging implements IMessaging, EventHandler { private RingBuffer m_ringBuffer; - private IStorageService m_storageService; + private IMessagesStore m_storageService; + private ISessionsStore m_sessionsStore; private ExecutorService m_executor; BatchEventProcessor m_eventProcessor; @@ -201,16 +203,19 @@ public class SimpleMessaging implements IMessaging, EventHandler { } private void processInit(Properties props) { - m_storageService = new HawtDBStorageService(); + HawtDBPersistentStore hawtStorage = new HawtDBPersistentStore(); + m_storageService = hawtStorage; m_storageService.initStore(); - List storedSubscriptions = m_storageService.listAllSubscriptions(); + m_sessionsStore = hawtStorage; + + List storedSubscriptions = m_sessionsStore.listAllSubscriptions(); subscriptions.init(storedSubscriptions); String passwdPath = props.getProperty("password_file"); String configPath = System.getProperty("moquette.path", null); IAuthenticator authenticator = new FileAuthenticator(configPath, passwdPath); - m_processor.init(subscriptions, m_storageService, authenticator); + m_processor.init(subscriptions, m_storageService, m_sessionsStore, authenticator); } diff --git a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageServiceTest.java b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBPersistentStoreTest.java similarity index 94% rename from broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageServiceTest.java rename to broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBPersistentStoreTest.java index 05df4878..e1bf0648 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageServiceTest.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBPersistentStoreTest.java @@ -29,16 +29,16 @@ import org.junit.Before; * * @author andrea */ -public class HawtDBStorageServiceTest { +public class HawtDBPersistentStoreTest { - HawtDBStorageService m_storageService; + HawtDBPersistentStore m_storageService; @Before public void setUp() throws Exception { File dbFile = new File(Server.STORAGE_FILE_PATH); assertFalse(String.format("The DB storagefile %s already exists", Server.STORAGE_FILE_PATH), dbFile.exists()); - m_storageService = new HawtDBStorageService(); + m_storageService = new HawtDBPersistentStore(); m_storageService.initStore(); } diff --git a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java index 0a2a91b0..02da3116 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java @@ -23,7 +23,8 @@ import java.util.List; import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; -import org.dna.mqtt.moquette.messaging.spi.IStorageService; +import org.dna.mqtt.moquette.messaging.spi.ISessionsStore; +import org.dna.mqtt.moquette.messaging.spi.IMessagesStore; import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent; import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription; import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.SubscriptionsStore; @@ -58,7 +59,8 @@ public class ProtocolProcessorTest { ConnectMessage connMsg; ProtocolProcessor m_processor; - IStorageService m_storageService; + IMessagesStore m_storageService; + ISessionsStore m_sessionStore; SubscriptionsStore subscriptions; AbstractMessage m_receivedMessage; MockAuthenticator m_mockAuthenticator; @@ -144,8 +146,9 @@ public class ProtocolProcessorTest { //sleep to let the messaging batch processor to process the initEvent Thread.sleep(300); - - m_storageService = new MemoryStorageService(); + MemoryStorageService memStorage = new MemoryStorageService(); + m_storageService = memStorage; + m_sessionStore = memStorage; //m_storageService.initStore(); Map users = new HashMap(); @@ -155,7 +158,7 @@ public class ProtocolProcessorTest { subscriptions = new SubscriptionsStore(); subscriptions.init(Collections.emptyList()); m_processor = new ProtocolProcessor(); - m_processor.init(subscriptions, m_storageService, m_mockAuthenticator); + m_processor.init(subscriptions, m_storageService, m_sessionStore, m_mockAuthenticator); } @Test @@ -263,7 +266,7 @@ public class ProtocolProcessorTest { //simulate a connect that register a clientID to an IoSession subs.init(Collections.emptyList()); - m_processor.init(subs, m_storageService, null); + m_processor.init(subs, m_storageService, m_sessionStore, null); ConnectMessage connectMessage = new ConnectMessage(); connectMessage.setProcotolVersion((byte)3); connectMessage.setClientID(FAKE_CLIENT_ID); @@ -302,7 +305,7 @@ public class ProtocolProcessorTest { //simulate a connect that register a clientID to an IoSession subs.init(Collections.emptyList()); - m_processor.init(subs, m_storageService, null); + m_processor.init(subs, m_storageService, m_sessionStore, null); MockReceiverChannel firstReceiverSession = new MockReceiverChannel(); ConnectMessage connectMessage = new ConnectMessage(); @@ -412,7 +415,7 @@ public class ProtocolProcessorTest { subs.init(Collections.emptyList()); //simulate a connect that register a clientID to an IoSession - m_processor.init(subs, m_storageService, null); + m_processor.init(subs, m_storageService, m_sessionStore, null); ConnectMessage connectMessage = new ConnectMessage(); connectMessage.setClientID(FAKE_PUBLISHER_ID); connectMessage.setProcotolVersion((byte)3); @@ -444,7 +447,7 @@ public class ProtocolProcessorTest { List inactiveSubscriptions = Arrays.asList(inactiveSub); when(mockedSubscriptions.matches(eq("/topic"))).thenReturn(inactiveSubscriptions); m_processor = new ProtocolProcessor(); - m_processor.init(mockedSubscriptions, m_storageService, null); + m_processor.init(mockedSubscriptions, m_storageService, m_sessionStore, null); //Exercise ByteBuffer buffer = ByteBuffer.allocate(5).put("Hello".getBytes()); @@ -464,7 +467,7 @@ public class ProtocolProcessorTest { List inactiveSubscriptions = Arrays.asList(inactiveSub); when(mockedSubscriptions.matches(eq("/topic"))).thenReturn(inactiveSubscriptions); m_processor = new ProtocolProcessor(); - m_processor.init(mockedSubscriptions, m_storageService, null); + m_processor.init(mockedSubscriptions, m_storageService, m_sessionStore, null); //Exercise ByteBuffer buffer = ByteBuffer.allocate(5).put("Hello".getBytes());