Fixed implementation for HawtDB storage of subscriptions

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();
 }

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();
+    }
 }

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;

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<Subscription> subscriptions = new ArrayList<Subscription>();
     private TreeNode subscriptions = new TreeNode(null);
-    
     private static final Logger LOG = LoggerFactory.getLogger(SubscriptionsStore.class);
-    
     //pesistent Map of clientID, list of Subscriptions
     private SortedIndex<String, List<Subscription>> 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<String, List<Subscription>> indexFactory = 
 //                new HashIndexFactory<String, List<Subscription>>();
-        BTreeIndexFactory<String, List<Subscription>> indexFactory = 
+        BTreeIndexFactory<String, List<Subscription>> indexFactory =
                 new BTreeIndexFactory<String, List<Subscription>>();
         indexFactory.setKeyCodec(StringCodec.INSTANCE);
-        
+
         m_persistent = indexFactory.openOrCreate(pageFile);
-        
+
         //reaload any subscriptions persisted
         LOG.debug("Reloading all stored subscriptions...");
-        for(Map.Entry<String, List<Subscription>> entry: m_persistent) {
+        for (Map.Entry<String, List<Subscription>> 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<Token> tokens = new ArrayList<Token>();
         try {
@@ -275,10 +290,11 @@ public class SubscriptionsStore {
         if (!m_persistent.containsKey(clientID)) {
             m_persistent.put(clientID, new ArrayList<Subscription>());
         }
-        
+
         List<Subscription> 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);
     }

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);

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();
+        }
     }