diff --git a/ChangeLog.txt b/ChangeLog.txt index 8fc6d1ca..9a0b89ea 100644 --- a/ChangeLog.txt +++ b/ChangeLog.txt @@ -2,6 +2,7 @@ Version 0.7: Fix issue #38, added aligned to MQTT 3.1.1 Refactored big event switch to @Annotation dispatcher Switched from HawtDB to MapDB + Fixed bug that didn't update client subscription on persistent store so that on server restart there where ghost subscriptions. Version 0.6: Fix issue #42, bad password file path composition diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/ISessionsStore.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/ISessionsStore.java index 6926aa0a..be1583d9 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/ISessionsStore.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/ISessionsStore.java @@ -16,6 +16,8 @@ package org.dna.mqtt.moquette.messaging.spi; import java.util.List; +import java.util.Set; + import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription; /** @@ -35,6 +37,11 @@ public interface ISessionsStore { */ void wipeSubscriptions(String sessionID); + /** + * Updates the subscriptions set for the clientID + * */ + void updateSubscriptions(String clientID, Set subscriptions); + List listAllSubscriptions(); /** 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 17dbf554..eefbf187 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 @@ -142,7 +142,12 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore { public void wipeSubscriptions(String clientID) { m_persistentSubscriptions.remove(clientID); } - + + @Override + public void updateSubscriptions(String clientID, Set subscriptions) { + m_persistentSubscriptions.put(clientID, subscriptions); + } + @Override public boolean contains(String clientID) { return m_persistentSubscriptions.containsKey(clientID); 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 f96305b8..81f17c63 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 @@ -19,7 +19,6 @@ import com.lmax.disruptor.EventHandler; import com.lmax.disruptor.RingBuffer; import com.lmax.disruptor.dsl.Disruptor; -import java.util.List; import java.util.Properties; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; @@ -30,7 +29,6 @@ import org.dna.mqtt.moquette.messaging.spi.IMessaging; 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; import org.dna.mqtt.moquette.messaging.spi.persistence.MapDBPersistentStore; import org.dna.mqtt.moquette.proto.messages.*; @@ -61,6 +59,7 @@ public class SimpleMessaging implements IMessaging, EventHandler { private ISessionsStore m_sessionsStore; private ExecutorService m_executor; + private Disruptor m_disruptor; private static SimpleMessaging INSTANCE; @@ -82,14 +81,14 @@ public class SimpleMessaging implements IMessaging, EventHandler { public void init(Properties configProps) { subscriptions = new SubscriptionsStore(); m_executor = Executors.newFixedThreadPool(1); - Disruptor disruptor = new Disruptor(ValueEvent.EVENT_FACTORY, 1024 * 32, m_executor); - /*Disruptor disruptor = new Disruptor(ValueEvent.EVENT_FACTORY, 1024 * 32, m_executor, + m_disruptor = new Disruptor(ValueEvent.EVENT_FACTORY, 1024 * 32, m_executor); + /*Disruptor m_disruptor = new Disruptor(ValueEvent.EVENT_FACTORY, 1024 * 32, m_executor, ProducerType.MULTI, new BusySpinWaitStrategy());*/ - disruptor.handleEventsWith(this); - disruptor.start(); + m_disruptor.handleEventsWith(this); + m_disruptor.start(); // Get the ring buffer from the Disruptor to be used for publishing. - m_ringBuffer = disruptor.getRingBuffer(); + m_ringBuffer = m_disruptor.getRingBuffer(); annotationSupport.processAnnotations(m_processor); processInit(configProps); @@ -127,6 +126,7 @@ public class SimpleMessaging implements IMessaging, EventHandler { boolean elapsed = !m_stopLatch.await(10, TimeUnit.SECONDS); LOG.debug("after m_stopLatch"); m_executor.shutdown(); + m_disruptor.shutdown(); if (elapsed) { LOG.error("Can't stop the server in 10 seconds"); } @@ -164,8 +164,9 @@ public class SimpleMessaging implements IMessaging, EventHandler { m_storageService.initStore(); - List storedSubscriptions = m_sessionsStore.listAllSubscriptions(); - subscriptions.init(storedSubscriptions); + //List storedSubscriptions = m_sessionsStore.listAllSubscriptions(); + //subscriptions.init(storedSubscriptions); + subscriptions.init(m_sessionsStore); String passwdPath = props.getProperty("password_file", ""); String configPath = System.getProperty("moquette.path", null); diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStore.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStore.java index 3462c10a..853c4b73 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStore.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStore.java @@ -16,12 +16,10 @@ package org.dna.mqtt.moquette.messaging.spi.impl.subscriptions; import java.text.ParseException; -import java.util.ArrayList; -import java.util.Collections; -import java.util.List; -import java.util.Queue; +import java.util.*; import java.util.concurrent.LinkedBlockingDeque; +import org.dna.mqtt.moquette.messaging.spi.ISessionsStore; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -31,7 +29,7 @@ import org.slf4j.LoggerFactory; * @author andrea */ public class SubscriptionsStore { - + public static interface IVisitor { void visit(TreeNode node); @@ -70,19 +68,21 @@ public class SubscriptionsStore { } private TreeNode subscriptions = new TreeNode(null); + private ISessionsStore m_sessionsStore; private static final Logger LOG = LoggerFactory.getLogger(SubscriptionsStore.class); /** * Initialize the subscription tree with the list of subscriptions. */ - public void init(List subscriptions) { + public void init(ISessionsStore sessionsStore) { LOG.debug("init invoked"); - + m_sessionsStore = sessionsStore; + List subscriptions = sessionsStore.listAllSubscriptions(); //reload any subscriptions persisted if (LOG.isDebugEnabled()) { LOG.debug("Reloading all stored subscriptions...subscription tree before {}", dumpTree()); } - + for (Subscription subscription : subscriptions) { LOG.debug("Re-subscribing {} to topic {}", subscription.getClientId(), subscription.getTopicFilter()); addDirect(subscription); @@ -169,18 +169,24 @@ public class SubscriptionsStore { */ public void removeForClient(String clientID) { subscriptions.removeClientSubscriptions(clientID); - + //TODO persist the update //remove from log all subscriptions // m_storageService.removeAllSubscriptions(clientID); } public void deactivate(String clientID) { subscriptions.deactivate(clientID); + //persist the update + Set subs = subscriptions.findAllByClientID(clientID); + m_sessionsStore.updateSubscriptions(clientID, subs); } public void activate(String clientID) { LOG.debug("Activating subscriptions for clientID <{}>", clientID); subscriptions.activate(clientID); + //persist the update + Set subs = subscriptions.findAllByClientID(clientID); + m_sessionsStore.updateSubscriptions(clientID, subs); } /** diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/TreeNode.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/TreeNode.java index d51ab8f3..d84eb67f 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/TreeNode.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/TreeNode.java @@ -15,15 +15,11 @@ */ package org.dna.mqtt.moquette.messaging.spi.impl.subscriptions; -import java.util.ArrayList; -import java.util.Collections; -import java.util.Comparator; -import java.util.List; -import java.util.Queue; +import java.util.*; import java.util.concurrent.LinkedBlockingQueue; class TreeNode { - + private class ClientIDComparator implements Comparator { public int compare(Subscription o1, Subscription o2) { @@ -181,6 +177,22 @@ class TreeNode { for (TreeNode child : m_children) { child.activate(clientID); } + } + /** + * @return the set of subscriptions for the given client. + * */ + Set findAllByClientID(String clientID) { + Set subs = new HashSet(); + for (Subscription s : m_subscriptions) { + if (s.clientId.equals(clientID)) { + subs.add(s); + } + } + //go deep + for (TreeNode child : m_children) { + subs.addAll(child.findAllByClientID(clientID)); + } + return subs; } } diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/persistence/MapDBPersistentStore.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/persistence/MapDBPersistentStore.java index a8123028..a975b6e1 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/persistence/MapDBPersistentStore.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/persistence/MapDBPersistentStore.java @@ -203,6 +203,12 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { m_db.commit(); } + @Override + public void updateSubscriptions(String clientID, Set subscriptions) { + m_persistentSubscriptions.put(clientID, subscriptions); + m_db.commit(); + } + public List listAllSubscriptions() { List allSubscriptions = new ArrayList(); for (Map.Entry> entry : m_persistentSubscriptions.entrySet()) { @@ -219,6 +225,7 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { public void close() { this.m_db.commit(); + LOG.debug("persisted subscriptions {}", m_persistentSubscriptions); this.m_db.close(); LOG.debug("closed disk storage"); } 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 79db84e1..5fc0b9f3 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 @@ -162,7 +162,7 @@ public class ProtocolProcessorTest { m_mockAuthenticator = new MockAuthenticator(users); subscriptions = new SubscriptionsStore(); - subscriptions.init(Collections.emptyList()); + subscriptions.init(new MemoryStorageService()); m_processor = new ProtocolProcessor(); m_processor.init(subscriptions, m_storageService, m_sessionStore, m_mockAuthenticator); } @@ -298,7 +298,7 @@ public class ProtocolProcessorTest { }; //simulate a connect that register a clientID to an IoSession - subs.init(Collections.emptyList()); + subs.init(new MemoryStorageService()); m_processor.init(subs, m_storageService, m_sessionStore, null); ConnectMessage connectMessage = new ConnectMessage(); connectMessage.setProcotolVersion((byte)3); @@ -342,7 +342,7 @@ public class ProtocolProcessorTest { }; //simulate a connect that register a clientID to an IoSession - subs.init(Collections.emptyList()); + subs.init(new MemoryStorageService()); m_processor.init(subs, m_storageService, m_sessionStore, null); MockReceiverChannel firstReceiverSession = new MockReceiverChannel(); @@ -372,7 +372,7 @@ public class ProtocolProcessorTest { m_processor.processPublish(m_session, msg); //Verify - Thread.sleep(100); //ugly but we dependend on the asynch that pull data from back disruptor + Thread.sleep(100); //ugly but we depend on the asynch that pull data from back disruptor PublishMessage pub2FirstSubscriber = (PublishMessage) firstReceiverSession.getMessage(); assertNotNull(pub2FirstSubscriber); String firstMessageContent = DebugUtils.payload2Str(pub2FirstSubscriber.getPayload()); @@ -458,7 +458,7 @@ public class ProtocolProcessorTest { } } }; - subs.init(Collections.emptyList()); + subs.init(new MemoryStorageService()); //simulate a connect that register a clientID to an IoSession m_processor.init(subs, m_storageService, m_sessionStore, null); diff --git a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStoreTest.java b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStoreTest.java index 26e1d9d6..da17a926 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStoreTest.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStoreTest.java @@ -22,6 +22,7 @@ import java.util.Arrays; import java.util.Collections; import java.util.List; +import org.dna.mqtt.moquette.messaging.spi.impl.MemoryStorageService; import org.dna.mqtt.moquette.proto.messages.AbstractMessage; import org.junit.After; import static org.junit.Assert.*; @@ -42,7 +43,7 @@ public class SubscriptionsStoreTest { @Before public void setUp() throws IOException { store = new SubscriptionsStore(); - store.init(Collections.emptyList()); + store.init(new MemoryStorageService()); } @Test @@ -232,7 +233,7 @@ public class SubscriptionsStoreTest { private void assertMatch(String subscription, String topic) { store = new SubscriptionsStore(); - store.init(Collections.emptyList()); + store.init(new MemoryStorageService()); Subscription sub = new Subscription("FAKE_CLI_ID_1", subscription, AbstractMessage.QOSType.MOST_ONE, false); store.add(sub); assertFalse(store.matches(topic).isEmpty()); @@ -240,7 +241,7 @@ public class SubscriptionsStoreTest { private void assertNotMatch(String subscription, String topic) { store = new SubscriptionsStore(); - store.init(Collections.emptyList()); + store.init(new MemoryStorageService()); Subscription sub = new Subscription("FAKE_CLI_ID_1", subscription, AbstractMessage.QOSType.MOST_ONE, false); store.add(sub); assertTrue(store.matches(topic).isEmpty()); @@ -325,7 +326,7 @@ public class SubscriptionsStoreTest { @Test public void removeSubscription_withDifferentClients_subscribedSameTopic() { SubscriptionsStore aStore = new SubscriptionsStore(); - aStore.init(Collections.emptyList()); + aStore.init(new MemoryStorageService()); //subscribe a not active clientID1 to /topic Subscription slashSub = new Subscription("FAKE_CLI_ID_1", "/topic", AbstractMessage.QOSType.MOST_ONE, false); aStore.add(slashSub); diff --git a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerLowlevelMessagesIntegrationTests.java b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerLowlevelMessagesIntegrationTests.java index 0eed28d7..32bf66d8 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerLowlevelMessagesIntegrationTests.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerLowlevelMessagesIntegrationTests.java @@ -64,8 +64,10 @@ public class ServerLowlevelMessagesIntegrationTests { @After public void tearDown() throws Exception { m_client.close(); - + LOG.debug("After raw client close"); + 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(Server.STORAGE_FILE_PATH); if (dbFile.exists()) { dbFile.delete(); @@ -111,11 +113,14 @@ public class ServerLowlevelMessagesIntegrationTests { //Execute m_client.sendMessage(connectMessage); long connectTime = System.currentTimeMillis(); - - //but after the 2 KEEP ALIVE timeut expires it gets fired, + + //but after the 2 KEEP ALIVE timeout expires it gets fired, //NB it's 1,5 * KEEP_ALIVE so 3 secs and some millis to propagate the message - Message msg = willSubscriber.receive(3100, TimeUnit.MILLISECONDS); + Message msg = willSubscriber.receive(3300, TimeUnit.MILLISECONDS); long willMessageReceiveTime = System.currentTimeMillis(); + if (msg == null) { + LOG.warn("testament message is null"); + } assertNotNull("the will message should be fired after keep alive!", msg); msg.ack(); //the will message hasn't to be received before the elapsing of Keep Alive timeout diff --git a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerRestartIntegrationTest.java b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerRestartIntegrationTest.java index dd5cc192..ed4f9e38 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerRestartIntegrationTest.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerRestartIntegrationTest.java @@ -49,7 +49,9 @@ public class ServerRestartIntegrationTest { startServer(); m_mqtt = new MQTT(); + m_mqtt.setHost("localhost", 1883); + m_mqtt.setCleanSession(false); } @After @@ -72,10 +74,8 @@ public class ServerRestartIntegrationTest { @Test public void checkRestartCleanSubscriptionTree() throws Exception { - //subscribe to /topic - m_mqtt.setHost("localhost", 1883); - m_mqtt.setCleanSession(false); m_mqtt.setClientId("Subscriber"); + //subscribe to /topic m_subscriber = m_mqtt.blockingConnection(); m_subscriber.connect(); Topic[] topics = new Topic[]{new Topic("/topic", QoS.AT_LEAST_ONCE)}; @@ -98,7 +98,7 @@ public class ServerRestartIntegrationTest { topics = new Topic[]{new Topic("/topic", QoS.EXACTLY_ONCE)}; m_subscriber.subscribe(topics); - //shoud be just one registration so a publisher receive one notification + //should be just one registration so a publisher receive one notification MQTT mqtt = new MQTT(); mqtt.setHost("localhost", 1883); mqtt.setClientId("Publisher"); @@ -113,4 +113,41 @@ public class ServerRestartIntegrationTest { //no more messages on the same topic will be received assertNull(m_subscriber.receive(1, TimeUnit.SECONDS)); } + + + @Test + public void checkDontPublishInactiveClientsAfterServerRestart() throws Exception { + m_mqtt.setClientId("SubPub"); + BlockingConnection conn = subscribeAndPublish("/topic"); + conn.disconnect(); + + //shutdown the server + m_server.stopServer(); + + //restart the server + m_server.startServer(); + + MQTT mqtt = new MQTT(); + mqtt.setHost("localhost", 1883); + mqtt.setClientId("Publisher"); + m_publisher = mqtt.blockingConnection(); + m_publisher.connect(); + m_publisher.publish("/topic", "Hello world MQTT!!".getBytes(), QoS.AT_MOST_ONCE, false); + } + + /** + * Connect subscribe to topic and publish on the same topic + * */ + private BlockingConnection subscribeAndPublish(String topic) throws Exception { + BlockingConnection conn = m_mqtt.blockingConnection(); + conn.connect(); + Topic[] topics = new Topic[]{new Topic(topic, QoS.AT_MOST_ONCE)}; + conn.subscribe(topics); + conn.publish(topic, "Hello world MQTT!!".getBytes(), QoS.AT_MOST_ONCE, false); + //read the message + Message msg = conn.receive(); + msg.ack(); + assertEquals("Hello world MQTT!!", new String(msg.getPayload())); + return conn; + } } \ No newline at end of file