anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Moved all HawtDB stuff into HawtDBStorageService

Browse Source
This commit is contained in:
selva.andre
2012-09-01 17:00:22 +02:00
parent ae8a2afb8b
commit a5d28b67e9
4 changed files with 39 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java
+2
View File
@@ -38,4 +38,6 @@ public interface IStorageService {
void removeAllSubscriptions(String clientID);
List<Subscription> retrieveAllSubscriptions();
void close();
}
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java
+31 -11
View File
@@ -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<String, List<Subscription>> 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);
}
}
}
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
+2 -27
View File
@@ -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 {
parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/DummyStorageService.java
+4
View File
@@ -55,4 +55,8 @@ public class DummyStorageService implements IStorageService {
public List<Subscription> retrieveAllSubscriptions() {
return Collections.EMPTY_LIST;
}
public void close() {
//To change body of implemented methods use File | Settings | File Templates.
}
}