From 7fb919837315dcb7cfb6f748a13bf91f2d868871 Mon Sep 17 00:00:00 2001 From: andsel Date: Sun, 15 Nov 2015 13:17:46 +0100 Subject: [PATCH] Fixed missed store/load of guid for StoredPublishEvent --- .../main/java/org/eclipse/moquette/spi/IMessagesStore.java | 6 ------ .../moquette/spi/impl/storage/StoredPublishEvent.java | 6 ++++++ .../moquette/spi/persistence/MapDBPersistentStore.java | 4 +++- .../eclipse/moquette/server/ServerIntegrationPahoTest.java | 5 +++-- 4 files changed, 12 insertions(+), 9 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 ebd7c0ab..fd36080b 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/IMessagesStore.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/IMessagesStore.java @@ -81,14 +81,8 @@ public interface IMessagesStore { */ List listMessagesInSession(Collection guids); -// void removeMessage(String clientID, Integer packetID); - void dropMessagesInSession(String clientID); -// void cleanTemporaryPublish(String clientID, int packetID); - -// void storeTemporaryPublish(PublishEvent evt, String clientID, int packetID); - /** * Return the next valid packetIdentifer for the given client session. * */ diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/storage/StoredPublishEvent.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/storage/StoredPublishEvent.java index 23af0c8c..b9613911 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/storage/StoredPublishEvent.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/storage/StoredPublishEvent.java @@ -33,6 +33,7 @@ public class StoredPublishEvent implements Serializable { String m_clientID; //Optional attribute, available only fo QoS 1 and 2 Integer m_msgID; + String m_guid; public StoredPublishEvent(PublishEvent wrapped) { m_topic = wrapped.getTopic(); @@ -45,6 +46,7 @@ public class StoredPublishEvent implements Serializable { m_message = new byte[buffer.remaining()]; buffer.get(m_message); buffer.rewind(); + m_guid = wrapped.getGuid(); } public String getTopic() { @@ -70,4 +72,8 @@ public class StoredPublishEvent implements Serializable { public Integer getMessageID() { return m_msgID; } + + public String getGuid() { + return m_guid; + } } 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 43928079..28c228d0 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 @@ -428,7 +428,9 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { byte[] message = evt.getMessage(); ByteBuffer bbmessage = ByteBuffer.wrap(message); //bbmessage.flip(); - return new PublishEvent(evt.getTopic(), evt.getQos(), + PublishEvent pubEvt = new PublishEvent(evt.getTopic(), evt.getQos(), bbmessage, evt.isRetain(), evt.getClientID(), evt.getMessageID()); + pubEvt.setGuid(evt.getGuid()); + return pubEvt; } } 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 fe70981c..c1985015 100644 --- a/broker/src/test/java/org/eclipse/moquette/server/ServerIntegrationPahoTest.java +++ b/broker/src/test/java/org/eclipse/moquette/server/ServerIntegrationPahoTest.java @@ -61,8 +61,9 @@ public class ServerIntegrationPahoTest { @Before public void setUp() throws Exception { String dbPath = IntegrationUtils.localMapDBPath(); - File dbFile = new File(dbPath); - assertFalse(String.format("The DB storagefile %s already exists", dbPath), dbFile.exists()); + IntegrationUtils.cleanPersistenceFile(dbPath); +// File dbFile = new File(dbPath); +// assertFalse(String.format("The DB storagefile %s already exists", dbPath), dbFile.exists()); startServer();