From 6ab413de68232320f561c1796b15b111bcaa3443 Mon Sep 17 00:00:00 2001 From: andsel Date: Wed, 30 Sep 2015 21:41:55 +0200 Subject: [PATCH] Fixed server shutdown, it doesn't flushed the MapDB to disk --- .../main/java/org/eclipse/moquette/server/Server.java | 1 + .../org/eclipse/moquette/spi/impl/SimpleMessaging.java | 4 ++++ .../moquette/spi/persistence/MapDBPersistentStore.java | 9 ++++++--- 3 files changed, 11 insertions(+), 3 deletions(-) diff --git a/broker/src/main/java/org/eclipse/moquette/server/Server.java b/broker/src/main/java/org/eclipse/moquette/server/Server.java index da4b7c25..6a2211bb 100644 --- a/broker/src/main/java/org/eclipse/moquette/server/Server.java +++ b/broker/src/main/java/org/eclipse/moquette/server/Server.java @@ -102,6 +102,7 @@ public class Server { public void stopServer() { LOG.info("Server stopping..."); m_acceptor.close(); + SimpleMessaging.getInstance().shutdown(); LOG.info("Server stopped"); } } diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/SimpleMessaging.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/SimpleMessaging.java index 98f768f8..291b79a1 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/SimpleMessaging.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/SimpleMessaging.java @@ -179,4 +179,8 @@ public class SimpleMessaging { return instance; } + + public void shutdown() { + this.m_storageService.close(); + } } 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 b3175fb8..9063c619 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 @@ -82,9 +82,7 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { try { tmpFile = new File(m_storePath); boolean fileNewlyCreated = tmpFile.createNewFile(); - if (!fileNewlyCreated) { - LOG.warn("File [{}] already exists", m_storePath); - } + LOG.info("Starting with {} [{}] db file", fileNewlyCreated ? "fresh" : "existing", m_storePath); } catch (IOException ex) { LOG.error(null, ex); throw new MQTTException("Can't create temp file for subscriptions storage [" + m_storePath + "]", ex); @@ -248,6 +246,7 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { m_persistentSubscriptions.put(clientID, clientSubscriptions); } + @Override public void addNewSubscription(Subscription newSubscription) { LOG.debug("addNewSubscription invoked with subscription {}", newSubscription); final String clientID = newSubscription.getClientId(); @@ -276,6 +275,7 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { } } + @Override public void wipeSubscriptions(String clientID) { m_persistentSubscriptions.remove(clientID); } @@ -310,6 +310,7 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { m_persistentSubscriptions.put(clientID, new HashSet()); } + @Override public void close() { if (this.m_db.isClosed()) { LOG.debug("already closed"); @@ -324,11 +325,13 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { } /*-------- QoS 2 storage management --------------*/ + @Override public void persistQoS2Message(String publishKey, PublishEvent evt) { LOG.debug("persistQoS2Message store pubKey: {}, evt: {}", publishKey, evt); m_qos2Store.put(publishKey, convertToStored(evt)); } + @Override public void removeQoS2Message(String publishKey) { LOG.debug("Removing stored Q0S2 message <{}>", publishKey); m_qos2Store.remove(publishKey);