Renamed session store management methods
This commit is contained in:
@@ -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<PublishEvent> retrievePersistedPublishes(String clientID);
|
||||
List<PublishEvent> 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);
|
||||
|
||||
|
||||
@@ -93,12 +93,12 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore {
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<PublishEvent> retrievePersistedPublishes(String clientID) {
|
||||
public List<PublishEvent> 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<PublishEvent> 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);
|
||||
}
|
||||
|
||||
|
||||
@@ -232,13 +232,16 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
|
||||
|
||||
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<PublishEvent> 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<PublishEvent> 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<ValueEvent> {
|
||||
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<ValueEvent> {
|
||||
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<ValueEvent> {
|
||||
subscriptions.removeForClient(clientID);
|
||||
|
||||
//remove also the messages stored of type QoS1/2
|
||||
m_messagesStore.cleanPersistedPublishes(clientID);
|
||||
m_messagesStore.dropMessagesInSession(clientID);
|
||||
}
|
||||
|
||||
@MQTTMessage(message = PublishMessage.class)
|
||||
|
||||
@@ -125,7 +125,7 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore {
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<PublishEvent> retrievePersistedPublishes(String clientID) {
|
||||
public List<PublishEvent> listMessagesInSession(String clientID) {
|
||||
List<StoredPublishEvent> 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<StoredPublishEvent> 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();
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user