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

Moved the part to manage Pusblish messages persistence to StorageService

Browse Source
This commit is contained in:
selva.andre
2012-09-01 16:18:09 +02:00
parent 2f9c957310
commit 2ed87b2c1a
3 changed files with 49 additions and 22 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
+8
View File
@@ -1,10 +1,12 @@
package org.dna.mqtt.moquette.messaging.spi;
import org.dna.mqtt.moquette.messaging.spi.impl.SimpleMessaging;
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.
@@ -20,4 +22,10 @@ public interface IStorageService {
Collection<StoredMessage> searchMatching(IMatchingCondition condition);
void storePublishForFuture(PublishEvent evt);
List<PublishEvent> retrivePersistedPublishes(String clientID);
void cleanPersistedPublishes(String clientID);
}
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java
+38
View File
@@ -3,6 +3,8 @@ 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.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.fusesource.hawtbuf.codec.StringCodec;
import org.fusesource.hawtdb.api.BTreeIndexFactory;
@@ -30,6 +32,8 @@ public class HawtDBStorageService implements IStorageService {
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;
public HawtDBStorageService(MultiIndexFactory multiIndexFactory) {
@@ -38,6 +42,12 @@ public class HawtDBStorageService implements IStorageService {
public void initStore() {
initRetainedStore();
//init the message store for QoS 1/2 messages in clean sessions
initPersistentMessageStore();
}
private void initRetainedStore() {
BTreeIndexFactory<String, StoredMessage> indexFactory = new BTreeIndexFactory<String, StoredMessage>();
indexFactory.setKeyCodec(StringCodec.INSTANCE);
@@ -45,6 +55,14 @@ public class HawtDBStorageService implements IStorageService {
}
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 storeRetained(String topic, byte[] message, AbstractMessage.QOSType qos) {
if (message.length == 0) {
//clean the message from topic
@@ -69,4 +87,24 @@ public class HawtDBStorageService implements IStorageService {
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);
}
}
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
+3 -22
View File
@@ -18,7 +18,6 @@ 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;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
import org.dna.mqtt.moquette.server.Constants;
import org.dna.mqtt.moquette.server.Server;
@@ -70,8 +69,6 @@ public class SimpleMessaging implements IMessaging, Runnable {
//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;
@@ -94,8 +91,6 @@ public class SimpleMessaging implements IMessaging, Runnable {
subscriptions.init(m_multiIndexFactory);
initInflightMessageStore();
//init the message store for QoS 1/2 messages in clean sessions
initPersistentMessageStore();
m_storageService = new HawtDBStorageService(m_multiIndexFactory);
m_storageService.initStore();
@@ -113,12 +108,6 @@ public class SimpleMessaging implements IMessaging, Runnable {
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;
@@ -297,15 +286,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));
}
@@ -354,7 +335,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 {
@@ -375,7 +356,7 @@ public class SimpleMessaging implements IMessaging, Runnable {
}
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;
}