From aa79fe3145a64fad65ba7ee77aa13a7b42680509 Mon Sep 17 00:00:00 2001 From: andrea Date: Fri, 30 Jan 2015 00:28:01 +0100 Subject: [PATCH] Renamed session store management methods --- .../eclipse/moquette/spi/IMessagesStore.java | 6 +++--- .../spi/impl/MemoryStorageService.java | 6 +++--- .../moquette/spi/impl/ProtocolProcessor.java | 19 +++++++++++-------- .../spi/persistence/MapDBPersistentStore.java | 6 +++--- .../spi/impl/ProtocolProcessorTest.java | 2 +- 5 files changed, 21 insertions(+), 18 deletions(-) diff --git a/broker/src/main/java/org/eclipse/moquette/spi/IMessagesStore.java b/broker/src/main/java/org/eclipse/moquette/spi/IMessagesStore.java index 8b60ddd1..73569aa1 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/IMessagesStore.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/IMessagesStore.java @@ -75,11 +75,11 @@ public interface IMessagesStore { * For QoS1 and QoS2 with clean session flag, this method return the list of * missed publish events while the client was disconnected. */ - List retrievePersistedPublishes(String clientID); + List listMessagesInSession(String clientID); - void cleanPersistedPublishMessage(String clientID, int messageID); + void removeMessageInSession(String clientID, int messageID); - void cleanPersistedPublishes(String clientID); + void dropMessagesInSession(String clientID); void cleanInFlight(String msgID); diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java index f8b98312..4f004e20 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java @@ -93,12 +93,12 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore { } @Override - public List retrievePersistedPublishes(String clientID) { + public List listMessagesInSession(String clientID) { return new ArrayList<>(m_persistentMessageStore.get(clientID)); } @Override - public void cleanPersistedPublishMessage(String clientID, int messageID) { + public void removeMessageInSession(String clientID, int messageID) { List events = m_persistentMessageStore.get(clientID); PublishEvent toRemoveEvt = null; for (PublishEvent evt : events) { @@ -111,7 +111,7 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore { } @Override - public void cleanPersistedPublishes(String clientID) { + public void dropMessagesInSession(String clientID) { m_persistentMessageStore.remove(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 d45fc3f1..6648ae75 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 @@ -232,13 +232,16 @@ class ProtocolProcessor implements EventHandler { if (!msg.isCleanSession()) { //force the republish of stored QoS1 and QoS2 - republishStored(msg.getClientID()); + republishStoredInSession(msg.getClientID()); } } - - private void republishStored(String clientID) { - LOG.trace("republishStored invoked"); - List publishedEvents = m_messagesStore.retrievePersistedPublishes(clientID); + + /** + * Republish QoS1 and QoS2 messages stored into the session for the clientID. + * */ + private void republishStoredInSession(String clientID) { + LOG.trace("republishStoredInSession invoked"); + List publishedEvents = m_messagesStore.listMessagesInSession(clientID); if (publishedEvents == null) { LOG.info("No stored messages for client <{}>", clientID); return; @@ -248,7 +251,7 @@ class ProtocolProcessor implements EventHandler { for (PublishEvent pubEvt : publishedEvents) { sendPublish(pubEvt.getClientID(), pubEvt.getTopic(), pubEvt.getQos(), pubEvt.getMessage(), false, pubEvt.getMessageID()); - m_messagesStore.cleanPersistedPublishMessage(clientID, pubEvt.getMessageID()); + m_messagesStore.removeMessageInSession(clientID, pubEvt.getMessageID()); } } @@ -257,7 +260,7 @@ class ProtocolProcessor implements EventHandler { String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); int messageID = msg.getMessageID(); //Remove the message from message store - m_messagesStore.cleanPersistedPublishMessage(clientID, messageID); + m_messagesStore.removeMessageInSession(clientID, messageID); } private void cleanSession(String clientID) { @@ -267,7 +270,7 @@ class ProtocolProcessor implements EventHandler { subscriptions.removeForClient(clientID); //remove also the messages stored of type QoS1/2 - m_messagesStore.cleanPersistedPublishes(clientID); + m_messagesStore.dropMessagesInSession(clientID); } @MQTTMessage(message = PublishMessage.class) 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 f2c965e0..4c27f29d 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 @@ -125,7 +125,7 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { } @Override - public List retrievePersistedPublishes(String clientID) { + public List listMessagesInSession(String clientID) { List storedEvts = m_persistentMessageStore.get(clientID); if (storedEvts == null) { return null; @@ -138,7 +138,7 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { } @Override - public void cleanPersistedPublishMessage(String clientID, int messageID) { + public void removeMessageInSession(String clientID, int messageID) { List events = m_persistentMessageStore.get(clientID); if (events == null) { return; @@ -154,7 +154,7 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { m_db.commit(); } - public void cleanPersistedPublishes(String clientID) { + public void dropMessagesInSession(String clientID) { m_persistentMessageStore.remove(clientID); m_db.commit(); } 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 756d07ce..f066bea0 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 @@ -504,7 +504,7 @@ public class ProtocolProcessorTest { m_processor.processConnect(m_session, connectMessage); //Verify no messages are still stored - assertTrue(m_storageService.retrievePersistedPublishes(FAKE_PUBLISHER_ID).isEmpty()); + assertTrue(m_storageService.listMessagesInSession(FAKE_PUBLISHER_ID).isEmpty()); } @Test