diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java index b6633a84..0ee6b843 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java @@ -38,4 +38,6 @@ public interface IStorageService { void removeAllSubscriptions(String clientID); List retrieveAllSubscriptions(); + + void close(); } diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java index da85628a..76736f97 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java @@ -1,27 +1,25 @@ package org.dna.mqtt.moquette.messaging.spi.impl; +import org.dna.mqtt.moquette.MQTTException; import org.dna.mqtt.moquette.messaging.spi.IMatchingCondition; import org.dna.mqtt.moquette.messaging.spi.IStorageService; -import org.dna.mqtt.moquette.messaging.spi.impl.events.CleanInFlightEvent; import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent; -import org.dna.mqtt.moquette.messaging.spi.impl.events.RemoveAllSubscriptionsEvent; -import org.dna.mqtt.moquette.messaging.spi.impl.events.RepublishEvent; import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.server.Server; import org.fusesource.hawtbuf.codec.StringCodec; -import org.fusesource.hawtdb.api.BTreeIndexFactory; -import org.fusesource.hawtdb.api.MultiIndexFactory; -import org.fusesource.hawtdb.api.PageFileFactory; -import org.fusesource.hawtdb.api.SortedIndex; +import org.fusesource.hawtdb.api.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import static org.dna.mqtt.moquette.messaging.spi.impl.SimpleMessaging.StoredMessage; - +import java.io.File; +import java.io.IOException; import java.util.ArrayList; import java.util.Collection; import java.util.List; import java.util.Map; +import static org.dna.mqtt.moquette.messaging.spi.impl.SimpleMessaging.StoredMessage; + /** * Implementation of IStorageService backed by HawtDB @@ -43,8 +41,22 @@ public class HawtDBStorageService implements IStorageService { //TODO rename into persistentSubscriptions private SortedIndex> m_persistent; - public HawtDBStorageService(MultiIndexFactory multiIndexFactory) { - m_multiIndexFactory = multiIndexFactory; + public HawtDBStorageService() { + String storeFile = Server.STORAGE_FILE_PATH; + + pageFactory = new PageFileFactory(); + File tmpFile; + try { + tmpFile = new File(storeFile); + tmpFile.createNewFile(); + } catch (IOException ex) { + LOG.error(null, ex); + throw new MQTTException("Can't create temp file for subscriptions storage [" + storeFile + "]", ex); + } + pageFactory.setFile(tmpFile); + pageFactory.open(); + PageFile pageFile = pageFactory.getPageFile(); + m_multiIndexFactory = new MultiIndexFactory(pageFile); } @@ -165,4 +177,12 @@ public class HawtDBStorageService implements IStorageService { } return allSubscriptions; } + + public void close() { + try { + pageFactory.close(); + } catch (IOException ex) { + LOG.error(null, ex); + } + } } diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java index 7bf1588c..e8b13637 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java @@ -59,31 +59,10 @@ public class SimpleMessaging implements IMessaging, Runnable { private INotifier m_notifier; - //TODO remove - private MultiIndexFactory m_multiIndexFactory; - private PageFileFactory pageFactory; - private IStorageService m_storageService; public SimpleMessaging() { - String storeFile = Server.STORAGE_FILE_PATH; - - pageFactory = new PageFileFactory(); - File tmpFile; - try { - tmpFile = new File(storeFile); - tmpFile.createNewFile(); - } catch (IOException ex) { - LOG.error(null, ex); - throw new MQTTException("Can't create temp file for subscriptions storage [" + storeFile + "]", ex); - } - pageFactory.setFile(tmpFile); - pageFactory.open(); - PageFile pageFile = pageFactory.getPageFile(); - m_multiIndexFactory = new MultiIndexFactory(pageFile); - - - m_storageService = new HawtDBStorageService(m_multiIndexFactory); + m_storageService = new HawtDBStorageService(); m_storageService.initStore(); subscriptions.init(m_storageService); @@ -329,11 +308,7 @@ public class SimpleMessaging implements IMessaging, Runnable { private void processClose() { LOG.debug("processClose invoked"); - try { - pageFactory.close(); - } catch (IOException ex) { - LOG.error(null, ex); - } + m_storageService.close(); } private void processRepublish(RepublishEvent evt) throws InterruptedException { diff --git a/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/DummyStorageService.java b/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/DummyStorageService.java index 594f4ab4..3be661e1 100644 --- a/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/DummyStorageService.java +++ b/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/DummyStorageService.java @@ -55,4 +55,8 @@ public class DummyStorageService implements IStorageService { public List retrieveAllSubscriptions() { return Collections.EMPTY_LIST; } + + public void close() { + //To change body of implemented methods use File | Settings | File Templates. + } }