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

Merge branch 'storage_service_extraction'

Browse Source
This commit is contained in:
selva.andre
2012-09-01 17:02:42 +02:00
parent 2c2539e457 8519b47dc2
commit 3c99be6d25
7 changed files with 353 additions and 124 deletions
Download Patch File Download Diff File Expand all files Collapse all files
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMatchingCondition.java
+8
View File
@@ -0,0 +1,8 @@
package org.dna.mqtt.moquette.messaging.spi;
/**
*/
public interface IMatchingCondition {
boolean match(String key);
}
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java
+43
View File
@@ -0,0 +1,43 @@
package org.dna.mqtt.moquette.messaging.spi;
import org.dna.mqtt.moquette.messaging.spi.impl.SimpleMessaging;
import org.dna.mqtt.moquette.messaging.spi.impl.Subscription;
import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import static org.dna.mqtt.moquette.messaging.spi.impl.SimpleMessaging.StoredMessage;
import java.util.Collection;
import java.util.List;
/**
* Defines the SPI to be implemented by a StorageService that handle persistence of messages and subscriptions.
*/
public interface IStorageService {
/**
* Used to initialize all persistent store structures
* */
void initStore();
void storeRetained(String topic, byte[] message, AbstractMessage.QOSType qos);
Collection<StoredMessage> searchMatching(IMatchingCondition condition);
void storePublishForFuture(PublishEvent evt);
List<PublishEvent> retrivePersistedPublishes(String clientID);
void cleanPersistedPublishes(String clientID);
void cleanInFlight(String msgID);
void addInFlight(PublishEvent evt, String publishKey);
void addNewSubscription(Subscription newSubscription, String clientID);
void removeAllSubscriptions(String clientID);
List<Subscription> retrieveAllSubscriptions();
void close();
}
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java
+187
View File
@@ -0,0 +1,187 @@
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.PublishEvent;
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.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
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
*/
public class HawtDBStorageService implements IStorageService {
private static final Logger LOG = LoggerFactory.getLogger(HawtDBStorageService.class);
private MultiIndexFactory m_multiIndexFactory;
private PageFileFactory pageFactory;
//maps clientID to the list of pending messages stored
private SortedIndex<String, List<PublishEvent>> m_persistentMessageStore;
private SortedIndex<String, StoredMessage> m_retainedStore;
//bind clientID+MsgID -> evt message published
private SortedIndex<String, PublishEvent> m_inflightStore;
//persistent Map of clientID, list of Subscriptions
private SortedIndex<String, List<Subscription>> m_persistentSubscriptions;
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);
}
public void initStore() {
initRetainedStore();
//init the message store for QoS 1/2 messages in clean sessions
initPersistentMessageStore();
initInflightMessageStore();
initPersistentSubscriptions();
}
private void initRetainedStore() {
BTreeIndexFactory<String, StoredMessage> indexFactory = new BTreeIndexFactory<String, StoredMessage>();
indexFactory.setKeyCodec(StringCodec.INSTANCE);
m_retainedStore = (SortedIndex<String, StoredMessage>) m_multiIndexFactory.openOrCreate("retained", indexFactory);
}
private void initPersistentMessageStore() {
BTreeIndexFactory<String, List<PublishEvent>> indexFactory = new BTreeIndexFactory<String, List<PublishEvent>>();
indexFactory.setKeyCodec(StringCodec.INSTANCE);
m_persistentMessageStore = (SortedIndex<String, List<PublishEvent>>) m_multiIndexFactory.openOrCreate("persistedMessages", indexFactory);
}
private void initPersistentSubscriptions() {
BTreeIndexFactory<String, List<Subscription>> indexFactory =
new BTreeIndexFactory<String, List<Subscription>>();
indexFactory.setKeyCodec(StringCodec.INSTANCE);
m_persistentSubscriptions = (SortedIndex<String, List<Subscription>>) m_multiIndexFactory.openOrCreate("subscriptions", indexFactory);
}
/**
* Initialize the message store used to handle the temporary storage of QoS 1,2
* messages in flight.
*/
private void initInflightMessageStore() {
BTreeIndexFactory<String, PublishEvent> indexFactory = new BTreeIndexFactory<String, PublishEvent>();
indexFactory.setKeyCodec(StringCodec.INSTANCE);
m_inflightStore = (SortedIndex<String, PublishEvent>) m_multiIndexFactory.openOrCreate("inflight", indexFactory);
}
public void storeRetained(String topic, byte[] message, AbstractMessage.QOSType qos) {
if (message.length == 0) {
//clean the message from topic
m_retainedStore.remove(topic);
} else {
//store the message to the topic
m_retainedStore.put(topic, new StoredMessage(message, qos));
}
}
public Collection<StoredMessage> searchMatching(IMatchingCondition condition) {
LOG.debug("searchMatching scanning all retained messages, presents are " + m_retainedStore.size());
List<StoredMessage> results = new ArrayList<StoredMessage>();
for (Map.Entry<String, StoredMessage> entry : m_retainedStore) {
StoredMessage storedMsg = entry.getValue();
if (condition.match(entry.getKey())) {
results.add(storedMsg);
}
}
return results;
}
public void storePublishForFuture(PublishEvent evt) {
List<PublishEvent> storedEvents;
String clientID = evt.getClientID();
if (!m_persistentMessageStore.containsKey(clientID)) {
storedEvents = new ArrayList<PublishEvent>();
} else {
storedEvents = m_persistentMessageStore.get(clientID);
}
storedEvents.add(evt);
m_persistentMessageStore.put(clientID, storedEvents);
}
public List<PublishEvent> retrivePersistedPublishes(String clientID) {
return m_persistentMessageStore.get(clientID);
}
public void cleanPersistedPublishes(String clientID) {
m_persistentMessageStore.remove(clientID);
}
public void cleanInFlight(String msgID) {
m_inflightStore.remove(msgID);
}
public void addInFlight(PublishEvent evt, String publishKey) {
m_inflightStore.put(publishKey, evt);
}
public void addNewSubscription(Subscription newSubscription, String clientID) {
if (!m_persistentSubscriptions.containsKey(clientID)) {
m_persistentSubscriptions.put(clientID, new ArrayList<Subscription>());
}
List<Subscription> subs = m_persistentSubscriptions.get(clientID);
subs.add(newSubscription);
m_persistentSubscriptions.put(clientID, subs);
}
public void removeAllSubscriptions(String clientID) {
m_persistentSubscriptions.remove(clientID);
}
public List<Subscription> retrieveAllSubscriptions() {
List<Subscription> allSubscriptions = new ArrayList<Subscription>();
for (Map.Entry<String, List<Subscription>> entry : m_persistentSubscriptions) {
allSubscriptions.addAll(entry.getValue());
}
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
+39 -102
View File
@@ -1,43 +1,43 @@
package org.dna.mqtt.moquette.messaging.spi.impl;
import java.io.File;
import java.io.IOException;
import java.io.Serializable;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import org.apache.mina.core.session.IoSession;
import org.dna.mqtt.moquette.MQTTException;
import org.dna.mqtt.moquette.messaging.spi.IMatchingCondition;
import org.dna.mqtt.moquette.messaging.spi.IMessaging;
import org.dna.mqtt.moquette.messaging.spi.INotifier;
import org.dna.mqtt.moquette.messaging.spi.IStorageService;
import org.dna.mqtt.moquette.messaging.spi.impl.SubscriptionsStore.Token;
import org.dna.mqtt.moquette.messaging.spi.impl.events.*;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
import org.dna.mqtt.moquette.server.Constants;
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.PageFile;
import org.fusesource.hawtdb.api.PageFileFactory;
import org.fusesource.hawtdb.api.SortedIndex;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
import java.io.Serializable;
import java.text.ParseException;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
/**
*
* @author andrea
*/
public class SimpleMessaging implements IMessaging, Runnable {
private static class StoredMessage implements Serializable {
//TODO probably move this
public static class StoredMessage implements Serializable {
QOSType m_qos;
byte[] m_payload;
private StoredMessage(byte[] message, QOSType qos) {
StoredMessage(byte[] message, QOSType qos) {
m_qos = qos;
m_payload = message;
}
@@ -58,64 +58,17 @@ public class SimpleMessaging implements IMessaging, Runnable {
private BlockingQueue<MessagingEvent> m_inboundQueue = new LinkedBlockingQueue<MessagingEvent>();
private INotifier m_notifier;
private MultiIndexFactory m_multiIndexFactory;
private PageFileFactory pageFactory;
private SortedIndex<String, StoredMessage> m_retainedStore;
//bind clientID+MsgID -> evt message published
private SortedIndex<String, PublishEvent> m_inflightStore;
//maps clientID to the list of pending messages stored
private SortedIndex<String, List<PublishEvent>> m_persistentMessageStore;
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);
subscriptions.init(m_multiIndexFactory);
initRetainedMessageStore();
initInflightMessageStore();
//init the message store for QoS 1/2 messages in clean sessions
initPersistentMessageStore();
m_storageService = new HawtDBStorageService();
m_storageService.initStore();
subscriptions.init(m_storageService);
}
private void initRetainedMessageStore() {
BTreeIndexFactory<String, StoredMessage> indexFactory = new BTreeIndexFactory<String, StoredMessage>();
indexFactory.setKeyCodec(StringCodec.INSTANCE);
// m_persistent = indexFactory.openOrCreate(pageFile);
m_retainedStore = (SortedIndex<String, StoredMessage>) m_multiIndexFactory.openOrCreate("retained", indexFactory);
}
/**
* Initialize the message store used to handle the temporary storage of QoS 1,2
* messages in flight.
*/
private void initInflightMessageStore() {
BTreeIndexFactory<String, PublishEvent> indexFactory = new BTreeIndexFactory<String, PublishEvent>();
indexFactory.setKeyCodec(StringCodec.INSTANCE);
m_inflightStore = (SortedIndex<String, PublishEvent>) m_multiIndexFactory.openOrCreate("inflight", indexFactory);
}
private void initPersistentMessageStore() {
BTreeIndexFactory<String, List<PublishEvent>> indexFactory = new BTreeIndexFactory<String, List<PublishEvent>>();
indexFactory.setKeyCodec(StringCodec.INSTANCE);
m_persistentMessageStore = (SortedIndex<String, List<PublishEvent>>) m_multiIndexFactory.openOrCreate("persistedMessages", indexFactory);
}
public void setNotifier(INotifier notifier) {
m_notifier= notifier;
}
@@ -258,7 +211,7 @@ public class SimpleMessaging implements IMessaging, Runnable {
processDisconnect((DisconnectEvent) evt);
} else if (evt instanceof CleanInFlightEvent) {
//remove the message from inflight storage
m_inflightStore.remove(((CleanInFlightEvent)evt).getMsgId());
m_storageService.cleanInFlight(((CleanInFlightEvent) evt).getMsgId());
} else if (evt instanceof RepublishEvent) {
processRepublish((RepublishEvent) evt);
}
@@ -281,7 +234,7 @@ public class SimpleMessaging implements IMessaging, Runnable {
if (qos == QOSType.LEAST_ONE) {
//store the temporary message
String publishKey = String.format("%s%d", evt.getClientID(), evt.getMessageID());
m_inflightStore.put(publishKey, evt);
m_storageService.addInFlight(evt, publishKey);
cleanEvt = new CleanInFlightEvent(publishKey);
}
@@ -293,15 +246,7 @@ public class SimpleMessaging implements IMessaging, Runnable {
//QoS 1 or 2
//if the target subscription is not clean session and is not connected => store it
if (!sub.isCleanSession() && !sub.isActive()) {
List<PublishEvent> storedEvents;
String clientID = evt.getClientID();
if (!m_persistentMessageStore.containsKey(clientID)) {
storedEvents = new ArrayList<PublishEvent>();
} else {
storedEvents = m_persistentMessageStore.get(clientID);
}
storedEvents.add(evt);
m_persistentMessageStore.put(clientID, storedEvents);
m_storageService.storePublishForFuture(evt);
}
m_notifier.notify(new NotifyEvent(sub.clientId, topic, qos, message, false));
}
@@ -313,13 +258,7 @@ public class SimpleMessaging implements IMessaging, Runnable {
}
if (retain) {
if (message.length == 0) {
//clean the message from topic
m_retainedStore.remove(topic);
} else {
//store the message to the topic
m_retainedStore.put(topic, new StoredMessage(message, qos));
}
m_storageService.storeRetained(topic, message, qos);
}
}
@@ -327,20 +266,22 @@ public class SimpleMessaging implements IMessaging, Runnable {
protected void processSubscribe(SubscribeEvent evt) {
LOG.debug("processSubscribe invoked");
Subscription newSubscription = evt.getSubscription();
String topic = newSubscription.getTopic();
final String topic = newSubscription.getTopic();
subscriptions.add(newSubscription);
//scans retained messages to be published to the new subscription
LOG.debug("Scanning all retained messages, presents are " + m_retainedStore.size());
for (Map.Entry<String, StoredMessage> entry : m_retainedStore) {
StoredMessage storedMsg = entry.getValue();
if (matchTopics(entry.getKey(), topic)) {
//fire the as retained the message
LOG.debug("Inserting NotifyEvent into outbound for topic " + topic + " entry " + entry.getKey());
m_notifier.notify(new NotifyEvent(newSubscription.clientId, topic, storedMsg.getQos(),
storedMsg.getPayload(), true));
Collection<StoredMessage> messages = m_storageService.searchMatching(new IMatchingCondition() {
public boolean match(String key) {
return matchTopics(key, topic); //To change body of implemented methods use File | Settings | File Templates.
}
});
for (StoredMessage storedMsg : messages) {
//fire the as retained the message
LOG.debug("Inserting NotifyEvent into outbound for topic " + topic);
m_notifier.notify(new NotifyEvent(newSubscription.clientId, topic, storedMsg.getQos(),
storedMsg.getPayload(), true));
}
}
@@ -354,7 +295,7 @@ public class SimpleMessaging implements IMessaging, Runnable {
subscriptions.removeForClient(evt.getClientID());
//remove also the messages stored of type QoS1/2
m_persistentMessageStore.remove(evt.getClientID());
m_storageService.cleanPersistedPublishes(evt.getClientID());
}
private void processDisconnect(DisconnectEvent evt) throws InterruptedException {
@@ -367,15 +308,11 @@ 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 {
List<PublishEvent> publishedEvents = m_persistentMessageStore.get(evt.getClientID());
List<PublishEvent> publishedEvents = m_storageService.retrivePersistedPublishes(evt.getClientID());
if (publishedEvents == null) {
return;
}
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStore.java
+13 -21
View File
@@ -7,6 +7,8 @@ import java.util.Map;
import java.util.Queue;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.LinkedBlockingQueue;
import org.dna.mqtt.moquette.messaging.spi.IStorageService;
import org.fusesource.hawtbuf.codec.StringCodec;
import org.fusesource.hawtdb.api.BTreeIndexFactory;
import org.fusesource.hawtdb.api.MultiIndexFactory;
@@ -205,28 +207,23 @@ public class SubscriptionsStore {
}
private TreeNode subscriptions = new TreeNode(null);
private static final Logger LOG = LoggerFactory.getLogger(SubscriptionsStore.class);
//persistent Map of clientID, list of Subscriptions
private SortedIndex<String, List<Subscription>> m_persistent;
private IStorageService m_storageService;
/**
* Initialize basic store structures, like the FS storage to maintain
* client's topics subscriptions
*/
public void init(MultiIndexFactory multiFactory) {
public void init(IStorageService storageService) {
LOG.debug("init invoked");
BTreeIndexFactory<String, List<Subscription>> indexFactory =
new BTreeIndexFactory<String, List<Subscription>>();
indexFactory.setKeyCodec(StringCodec.INSTANCE);
m_persistent = (SortedIndex<String, List<Subscription>>) multiFactory.openOrCreate("subscriptions", indexFactory);
m_storageService = storageService;
//reload any subscriptions persisted
LOG.debug("Reloading all stored subscriptions...");
for (Map.Entry<String, List<Subscription>> entry : m_persistent) {
for (Subscription subscription : entry.getValue()) {
LOG.debug("Re-subscribing " + subscription.getClientId() + " to topic " + subscription.getTopic());
addDirect(subscription);
}
for (Subscription subscription : m_storageService.retrieveAllSubscriptions()) {
LOG.debug("Re-subscribing " + subscription.getClientId() + " to topic " + subscription.getTopic());
addDirect(subscription);
}
LOG.debug("Finished loading");
}
@@ -269,16 +266,11 @@ public class SubscriptionsStore {
//log the subscription
String clientID = newSubscription.getClientId();
if (!m_persistent.containsKey(clientID)) {
m_persistent.put(clientID, new ArrayList<Subscription>());
}
List<Subscription> subs = m_persistent.get(clientID);
subs.add(newSubscription);
m_persistent.put(clientID, subs);
m_storageService.addNewSubscription(newSubscription, clientID);
// pageFile.flush();
}
public void removeSubscription(String topic, String clientID) {
TreeNode matchNode = findMatchingNode(topic);
@@ -303,7 +295,7 @@ public class SubscriptionsStore {
subscriptions.removeClientSubscriptions(clientID);
//remove from log all subscriptions
m_persistent.remove(clientID);
m_storageService.removeAllSubscriptions(clientID);
}
public void disconnect(String clientID) {
parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/DummyStorageService.java
+62
View File
@@ -0,0 +1,62 @@
package org.dna.mqtt.moquette.messaging.spi.impl;
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.PublishEvent;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
/**
*/
public class DummyStorageService implements IStorageService {
public void initStore() {
//To change body of implemented methods use File | Settings | File Templates.
}
public void storeRetained(String topic, byte[] message, AbstractMessage.QOSType qos) {
//To change body of implemented methods use File | Settings | File Templates.
}
public Collection<SimpleMessaging.StoredMessage> searchMatching(IMatchingCondition condition) {
return null; //To change body of implemented methods use File | Settings | File Templates.
}
public void storePublishForFuture(PublishEvent evt) {
//To change body of implemented methods use File | Settings | File Templates.
}
public List<PublishEvent> retrivePersistedPublishes(String clientID) {
return null; //To change body of implemented methods use File | Settings | File Templates.
}
public void cleanPersistedPublishes(String clientID) {
//To change body of implemented methods use File | Settings | File Templates.
}
public void cleanInFlight(String msgID) {
//To change body of implemented methods use File | Settings | File Templates.
}
public void addInFlight(PublishEvent evt, String publishKey) {
//To change body of implemented methods use File | Settings | File Templates.
}
public void addNewSubscription(Subscription newSubscription, String clientID) {
//To change body of implemented methods use File | Settings | File Templates.
}
public void removeAllSubscriptions(String clientID) {
//To change body of implemented methods use File | Settings | File Templates.
}
public List<Subscription> retrieveAllSubscriptions() {
return Collections.EMPTY_LIST;
}
public void close() {
//To change body of implemented methods use File | Settings | File Templates.
}
}
parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStoreTest.java
+1 -1
View File
@@ -40,7 +40,7 @@ public class SubscriptionsStoreTest {
pageFactory.open();
PageFile pageFile = pageFactory.getPageFile();
MultiIndexFactory multiIndexFactory = new MultiIndexFactory(pageFile);
store.init(multiIndexFactory);
store.init(new DummyStorageService());
}
@After