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

Moved the fling messages part persistence into StorageService

Browse Source
This commit is contained in:
selva.andre
2012-09-01 16:28:03 +02:00
parent 2ed87b2c1a
commit b092d15c12
3 changed files with 36 additions and 31 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
+3
View File
@@ -28,4 +28,7 @@ public interface IStorageService {
void cleanPersistedPublishes(String clientID);
void cleanInFlight(String msgID);
void addInFlight(PublishEvent evt, String publishKey);
}
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java
+22
View File
@@ -2,6 +2,7 @@ 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.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;
@@ -35,6 +36,8 @@ public class HawtDBStorageService implements IStorageService {
//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;
public HawtDBStorageService(MultiIndexFactory multiIndexFactory) {
m_multiIndexFactory = multiIndexFactory;
@@ -63,6 +66,17 @@ public class HawtDBStorageService implements IStorageService {
m_persistentMessageStore = (SortedIndex<String, List<PublishEvent>>) m_multiIndexFactory.openOrCreate("persistedMessages", 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
@@ -107,4 +121,12 @@ public class HawtDBStorageService implements IStorageService {
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);
}
}
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
+11 -31
View File
@@ -1,15 +1,5 @@
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.Collection;
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;
@@ -21,15 +11,21 @@ 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
@@ -67,9 +63,6 @@ public class SimpleMessaging implements IMessaging, Runnable {
private MultiIndexFactory m_multiIndexFactory;
private PageFileFactory pageFactory;
//bind clientID+MsgID -> evt message published
private SortedIndex<String, PublishEvent> m_inflightStore;
private IStorageService m_storageService;
public SimpleMessaging() {
@@ -90,25 +83,12 @@ public class SimpleMessaging implements IMessaging, Runnable {
m_multiIndexFactory = new MultiIndexFactory(pageFile);
subscriptions.init(m_multiIndexFactory);
initInflightMessageStore();
m_storageService = new HawtDBStorageService(m_multiIndexFactory);
m_storageService.initStore();
}
/**
* 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 setNotifier(INotifier notifier) {
m_notifier= notifier;
}
@@ -251,7 +231,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);
}
@@ -274,7 +254,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);
}