diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java index c094bc14..32beebba 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java @@ -27,4 +27,6 @@ public interface IMessaging { * was subscribed */ void removeSubscriptions(String clientID); + + void close(); } 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 48a1f323..d46fd5be 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 @@ -58,4 +58,10 @@ public class SimpleMessaging implements IMessaging { subscriptions.removeForClient(clientID); rwLock.writeLock().unlock(); } + + public void close() { + rwLock.writeLock().lock(); + subscriptions.close(); + rwLock.writeLock().unlock(); + } } diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/Subscription.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/Subscription.java index da4b27d4..83fab74e 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/Subscription.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/Subscription.java @@ -1,5 +1,6 @@ package org.dna.mqtt.moquette.messaging.spi.impl; +import java.io.Serializable; import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; /** @@ -9,7 +10,7 @@ import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; * * @author andrea */ -public class Subscription { +public class Subscription implements Serializable { QOSType requestedQos; String clientId; diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStore.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStore.java index d87e39df..a6e3b729 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStore.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStore.java @@ -9,8 +9,10 @@ import java.util.Map; import java.util.Queue; import java.util.concurrent.LinkedBlockingDeque; import java.util.concurrent.LinkedBlockingQueue; +import java.util.logging.Level; import org.dna.mqtt.moquette.MQTTException; import org.dna.mqtt.moquette.server.MQTTHandler; +import org.dna.mqtt.moquette.server.Server; import org.fusesource.hawtbuf.codec.StringCodec; import org.fusesource.hawtdb.api.BTreeIndexFactory; import org.fusesource.hawtdb.api.PageFile; @@ -197,41 +199,46 @@ public class SubscriptionsStore { } // private List subscriptions = new ArrayList(); private TreeNode subscriptions = new TreeNode(null); - private static final Logger LOG = LoggerFactory.getLogger(SubscriptionsStore.class); - //pesistent Map of clientID, list of Subscriptions private SortedIndex> m_persistent; - private PageFile pageFile; +// private PageFile pageFile; + private PageFileFactory factory; /** * Initialize basic store structures, like the FS storage to maintain * client's topics subscriptions */ public void init() { - PageFileFactory factory = new PageFileFactory(); + /*StringBuilder storeFile = new StringBuilder(); + storeFile.append(System.getProperty("user.home")) + .append(File.separator).append("moquette_store.hawtdb");*/ + String storeFile = Server.STORAGE_FILE_PATH; + + factory = new PageFileFactory(); File tmpFile; try { - tmpFile = File.createTempFile("hawtdb", "test"); + tmpFile = new File(storeFile.toString()); + tmpFile.createNewFile(); } catch (IOException ex) { LOG.error(null, ex); - throw new MQTTException("Can't create temp file for subscriptions storage", ex); + throw new MQTTException("Can't create temp file for subscriptions storage [" + storeFile + "]", ex); } factory.setFile(/*new File("mydb.dat")*/tmpFile); factory.open(); - pageFile = factory.getPageFile(); + PageFile pageFile = factory.getPageFile(); // HashIndexFactory> indexFactory = // new HashIndexFactory>(); - BTreeIndexFactory> indexFactory = + BTreeIndexFactory> indexFactory = new BTreeIndexFactory>(); indexFactory.setKeyCodec(StringCodec.INSTANCE); - + m_persistent = indexFactory.openOrCreate(pageFile); - + //reaload any subscriptions persisted LOG.debug("Reloading all stored subscriptions..."); - for(Map.Entry> entry: m_persistent) { + for (Map.Entry> entry : m_persistent) { for (Subscription subscription : entry.getValue()) { addDirect(subscription); } @@ -239,7 +246,15 @@ public class SubscriptionsStore { LOG.debug("Finished loading"); } - + public void close() { + try { + factory.close(); + } catch (IOException ex) { + //TODO handle probably firing + LOG.error(null, ex); + } + } + protected void addDirect(Subscription newSubscription) { List tokens = new ArrayList(); try { @@ -275,10 +290,11 @@ public class SubscriptionsStore { if (!m_persistent.containsKey(clientID)) { m_persistent.put(clientID, new ArrayList()); } - + List subs = m_persistent.get(clientID); subs.add(newSubscription); - pageFile.flush(); + m_persistent.put(clientID, subs); +// pageFile.flush(); } /** @@ -286,7 +302,7 @@ public class SubscriptionsStore { */ public void removeForClient(String clientID) { subscriptions.removeClientSubscriptions(clientID); - + //remove from log all subscriptions m_persistent.remove(clientID); } diff --git a/parser/src/main/java/org/dna/mqtt/moquette/server/Server.java b/parser/src/main/java/org/dna/mqtt/moquette/server/Server.java index c0cb8307..0326798c 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/server/Server.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/server/Server.java @@ -1,5 +1,6 @@ package org.dna.mqtt.moquette.server; +import java.io.File; import java.io.IOException; import java.net.InetSocketAddress; import org.apache.mina.core.service.IoAcceptor; @@ -36,7 +37,10 @@ public class Server { public static final int PORT = 9191; public static final int DEFAULT_CONNECT_TIMEOUT = 10; + public static final String STORAGE_FILE_PATH = System.getProperty("user.home") + + File.separator + "moquette_store.hawtdb"; private IoAcceptor m_acceptor; + SimpleMessaging messaging; public static void main(String[] args) throws IOException { new Server().startServer(); @@ -64,7 +68,7 @@ public class Server { m_acceptor.getFilterChain().addLast( "codec", new ProtocolCodecFilter(encoder, decoder)); MQTTHandler handler = new MQTTHandler(); - SimpleMessaging messaging = new SimpleMessaging(); + messaging = new SimpleMessaging(); //TODO fix this hugly wiring handler.setMessaging(messaging); messaging.setNotifier(handler); @@ -89,6 +93,9 @@ public class Server { protected void stopServer() { LOG.info("Server stopping..."); + + messaging.close(); + for(IoSession session: m_acceptor.getManagedSessions().values()) { if(session.isConnected() && !session.isClosing()){ session.close(false); diff --git a/parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java b/parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java index 2cc83a3a..8f38e4eb 100644 --- a/parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java +++ b/parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java @@ -1,5 +1,6 @@ package org.dna.mqtt.moquette.server; +import java.io.File; import java.io.IOException; import org.dna.mqtt.moquette.client.Client; import org.dna.mqtt.moquette.client.IPublishCallback; @@ -30,6 +31,10 @@ public class ServerIntegrationTest { @After public void tearDown() { server.stopServer(); + File dbFile = new File(Server.STORAGE_FILE_PATH); + if (dbFile.exists()) { + dbFile.delete(); + } } diff --git a/tools_scripts/hawtdb_play.groovy b/tools_scripts/hawtdb_play.groovy new file mode 100644 index 00000000..17efe18e --- /dev/null +++ b/tools_scripts/hawtdb_play.groovy @@ -0,0 +1,65 @@ +@Grab(group='org.fusesource.hawtdb', module='hawtdb', version='1.6') + +import org.fusesource.hawtbuf.codec.StringCodec +import org.fusesource.hawtdb.api.BTreeIndexFactory +import org.fusesource.hawtdb.api.PageFile +import org.fusesource.hawtdb.api.PageFileFactory +import org.fusesource.hawtdb.api.SortedIndex + + + +File tmpFile = new File(System.getProperty("user.home") + File.separator + "test_hawtdb.dat") +//tmpFile.createNewFile() + +def writeSomeData(tmpFile) { + def factory = new PageFileFactory() + factory.file = tmpFile + factory.open() + def pageFile = factory.getPageFile() + + BTreeIndexFactory indexFactory = new BTreeIndexFactory() + indexFactory.setKeyCodec(StringCodec.INSTANCE) + + SortedIndex writeIndex = indexFactory.openOrCreate(pageFile) + println "Hawt opened, try to read" + println "Size: ${writeIndex.size()}" + + println "write something" + writeIndex.put("titolo", "capucetto rosso") + pageFile.flush() + + println "check after write content: ${writeIndex.get("titolo")}, size ${writeIndex.size()}" + + //pageFile.close() + factory.close() +} + + +def readAllFileContents(tmpFile) { + def factory = new PageFileFactory() + factory.file = tmpFile + factory.open() + def pageFile = factory.getPageFile() + + BTreeIndexFactory indexFactory = new BTreeIndexFactory() + indexFactory.setKeyCodec(StringCodec.INSTANCE) + + SortedIndex readingIndex = indexFactory.openOrCreate(pageFile) + + println "read something, size ${readingIndex.size()}" + for (def entry : readingIndex) { + println "read k, v -> ${entry.key}, ${entry.value} " + } + + factory.close() +} + + +writeSomeData(tmpFile) +readAllFileContents(tmpFile) + + + + + + diff --git a/tools_scripts/hawtdb_raw_access.groovy b/tools_scripts/hawtdb_raw_access.groovy new file mode 100644 index 00000000..75579f46 --- /dev/null +++ b/tools_scripts/hawtdb_raw_access.groovy @@ -0,0 +1,45 @@ +@Grab(group='org.fusesource.hawtdb', module='hawtdb', version='1.6') + +import org.fusesource.hawtbuf.codec.StringCodec +import org.fusesource.hawtdb.api.BTreeIndexFactory +import org.fusesource.hawtdb.api.PageFile +import org.fusesource.hawtdb.api.PageFileFactory +import org.fusesource.hawtdb.api.SortedIndex +import org.fusesource.hawtbuf.Buffer + + + +File tmpFile = new File(System.getProperty("user.home") + File.separator + "test_hawtdb.dat") +//tmpFile.createNewFile() + +def writeSomeData(tmpFile) { + def factory = new PageFileFactory() + factory.file = tmpFile + factory.open() + def pageFile = factory.getPageFile() + int pageIdx = pageFile.alloc() + pageFile.write(pageIdx, new Buffer("Test".getBytes())) + println "writed at page index ${pageIdx}" + pageFile.flush() + pageFile.close() +} + + +def readTest(tmpFile) { + def factory = new PageFileFactory() + factory.file = tmpFile + factory.open() + def pageFile = factory.getPageFile() + println "page size is: " + pageFile.pageSize + Buffer b = new Buffer(512) + pageFile.read(0, b) + println "content try read " + b + + pageFile.close() +} + + +writeSomeData(tmpFile) +readTest(tmpFile) + +