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

Separated the interfaces to persist messages from the one that persists sessions

Browse Source
This commit is contained in:
selva.andre
2014-06-30 23:36:00 +02:00
parent d8ec0f2e94
commit 08e12494ec
9 changed files with 59 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ChangeLog.txt
+2
View File
@@ -5,6 +5,8 @@ Version 0.6:
Integrated SSL support (issue #41)
Fixed issue #40, check Logging inside the Felix OSGi
Fixed issue #46, implement correct behaviour for Will stuff
Splitted stores to persist messages and sessions into different interfaces
Renamed some methods in sessions store
Version 0.5:
Fix issue 20 (no flush of message on PINGREQ)
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java → broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessagesStore.java
+3 -3
View File
@@ -18,15 +18,15 @@ package org.dna.mqtt.moquette.messaging.spi;
import java.nio.ByteBuffer;
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.HawtDBStorageService.StoredMessage;
import static org.dna.mqtt.moquette.messaging.spi.impl.HawtDBPersistentStore.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.
* Defines the SPI to be implemented by a StorageService that handle persistence of messages
*/
public interface IStorageService extends IPersistentSubscriptionStore {
public interface IMessagesStore {
/**
* Used to initialize all persistent store structures
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IPersistentSubscriptionStore.java → broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/ISessionsStore.java
+1 -1
View File
@@ -23,7 +23,7 @@ import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription;
*
* @author andrea
*/
public interface IPersistentSubscriptionStore {
public interface ISessionsStore {
/**
* Add a new subscription to the session
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java → broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBPersistentStore.java
+6 -5
View File
@@ -22,7 +22,8 @@ import java.nio.ByteBuffer;
import java.util.*;
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.ISessionsStore;
import org.dna.mqtt.moquette.messaging.spi.IMessagesStore;
import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.storage.StoredPublishEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription;
@@ -35,9 +36,9 @@ import org.slf4j.LoggerFactory;
/**
* Implementation of IStorageService backed by HawtDB
* Implementation of IMessagesStore and ISessionsStore backed by HawtDB
*/
public class HawtDBStorageService implements IStorageService {
public class HawtDBPersistentStore implements IMessagesStore, ISessionsStore {
public static class StoredMessage implements Serializable {
AbstractMessage.QOSType m_qos;
@@ -64,7 +65,7 @@ public class HawtDBStorageService implements IStorageService {
}
private static final Logger LOG = LoggerFactory.getLogger(HawtDBStorageService.class);
private static final Logger LOG = LoggerFactory.getLogger(HawtDBPersistentStore.class);
private MultiIndexFactory m_multiIndexFactory;
private PageFileFactory pageFactory;
@@ -81,7 +82,7 @@ public class HawtDBStorageService implements IStorageService {
//persistent Map of clientID, set of Subscriptions
private SortedIndex<String, Set<Subscription>> m_persistentSubscriptions;
public HawtDBStorageService() {
public HawtDBPersistentStore() {
String storeFile = Server.STORAGE_FILE_PATH;
pageFactory = new PageFileFactory();
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java
+9 -8
View File
@@ -18,7 +18,7 @@ package org.dna.mqtt.moquette.messaging.spi.impl;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import org.dna.mqtt.moquette.messaging.spi.IMatchingCondition;
import org.dna.mqtt.moquette.messaging.spi.IStorageService;
import org.dna.mqtt.moquette.messaging.spi.IMessagesStore;
import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
@@ -29,15 +29,16 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.dna.mqtt.moquette.messaging.spi.ISessionsStore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
*/
public class MemoryStorageService implements IStorageService {
public class MemoryStorageService implements IMessagesStore, ISessionsStore {
private Map<String, Set<Subscription>> m_persistentSubscriptions = new HashMap<String, Set<Subscription>>();
private Map<String, HawtDBStorageService.StoredMessage> m_retainedStore = new HashMap<String, HawtDBStorageService.StoredMessage>();
private Map<String, HawtDBPersistentStore.StoredMessage> m_retainedStore = new HashMap<String, HawtDBPersistentStore.StoredMessage>();
//TODO move in a multimap because only Qos1 and QoS2 are stored here and they have messageID(key of secondary map)
private Map<String, List<PublishEvent>> m_persistentMessageStore = new HashMap<String, List<PublishEvent>>();
private Map<String, PublishEvent> m_inflightStore = new HashMap<String, PublishEvent>();
@@ -57,17 +58,17 @@ public class MemoryStorageService implements IStorageService {
//store the message to the topic
byte[] raw = new byte[message.remaining()];
message.get(raw);
m_retainedStore.put(topic, new HawtDBStorageService.StoredMessage(raw, qos, topic));
m_retainedStore.put(topic, new HawtDBPersistentStore.StoredMessage(raw, qos, topic));
}
}
public Collection<HawtDBStorageService.StoredMessage> searchMatching(IMatchingCondition condition) {
public Collection<HawtDBPersistentStore.StoredMessage> searchMatching(IMatchingCondition condition) {
LOG.debug("searchMatching scanning all retained messages, presents are {}", m_retainedStore.size());
List<HawtDBStorageService.StoredMessage> results = new ArrayList<HawtDBStorageService.StoredMessage>();
List<HawtDBPersistentStore.StoredMessage> results = new ArrayList<HawtDBPersistentStore.StoredMessage>();
for (Map.Entry<String, HawtDBStorageService.StoredMessage> entry : m_retainedStore.entrySet()) {
HawtDBStorageService.StoredMessage storedMsg = entry.getValue();
for (Map.Entry<String, HawtDBPersistentStore.StoredMessage> entry : m_retainedStore.entrySet()) {
HawtDBPersistentStore.StoredMessage storedMsg = entry.getValue();
if (condition.match(entry.getKey())) {
results.add(storedMsg);
}
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java
+12 -7
View File
@@ -27,7 +27,8 @@ import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import org.dna.mqtt.moquette.messaging.spi.IMatchingCondition;
import org.dna.mqtt.moquette.messaging.spi.IStorageService;
import org.dna.mqtt.moquette.messaging.spi.ISessionsStore;
import org.dna.mqtt.moquette.messaging.spi.IMessagesStore;
import org.dna.mqtt.moquette.messaging.spi.impl.events.MessagingEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.events.OutputMessagingEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.events.PubAckEvent;
@@ -98,7 +99,8 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
private Map<String, ConnectionDescriptor> m_clientIDs = new HashMap<String, ConnectionDescriptor>();
private SubscriptionsStore subscriptions;
private IStorageService m_storageService;
private IMessagesStore m_storageService;
private ISessionsStore m_sessionsStore;
private IAuthenticator m_authenticator;
//maps clientID to Will testament, if specified on CONNECT
private Map<String, WillMessage> m_willStore = new HashMap<String, WillMessage>();
@@ -114,14 +116,17 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
* clients subscriptions.
* @param storageService the persistent store to use for save/load of messages
* for QoS1 and QoS2 handling.
* @param sessionsStore the clients sessions store, used to persist subscriptions.
* @param authenticator the authenticator used in connect messages
*/
void init(SubscriptionsStore subscriptions, IStorageService storageService,
void init(SubscriptionsStore subscriptions, IMessagesStore storageService,
ISessionsStore sessionsStore,
IAuthenticator authenticator) {
//m_clientIDs = clientIDs;
this.subscriptions = subscriptions;
m_authenticator = authenticator;
m_storageService = storageService;
m_sessionsStore = sessionsStore;
//init the output ringbuffer
m_executor = Executors.newFixedThreadPool(1);
@@ -246,7 +251,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
private void cleanSession(String clientID) {
LOG.info("cleaning old saved subscriptions for client <{}>", clientID);
//remove from log all subscriptions
m_storageService.wipeSubscriptions(clientID);
m_sessionsStore.wipeSubscriptions(clientID);
subscriptions.removeForClient(clientID);
//remove also the messages stored of type QoS1/2
@@ -537,17 +542,17 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
newSubscription.getClientId(), topic,
AbstractMessage.QOSType.formatQoS(newSubscription.getRequestedQos()));
String clientID = newSubscription.getClientId();
m_storageService.addNewSubscription(newSubscription, clientID);
m_sessionsStore.addNewSubscription(newSubscription, clientID);
subscriptions.add(newSubscription);
//scans retained messages to be published to the new subscription
Collection<HawtDBStorageService.StoredMessage> messages = m_storageService.searchMatching(new IMatchingCondition() {
Collection<HawtDBPersistentStore.StoredMessage> messages = m_storageService.searchMatching(new IMatchingCondition() {
public boolean match(String key) {
return SubscriptionsStore.matchTopics(key, topic);
}
});
for (HawtDBStorageService.StoredMessage storedMsg : messages) {
for (HawtDBPersistentStore.StoredMessage storedMsg : messages) {
//fire the as retained the message
LOG.debug("send publish message for topic {}", topic);
sendPublish(newSubscription.getClientId(), storedMsg.getTopic(), storedMsg.getQos(), storedMsg.getPayload(), true);
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
+10 -5
View File
@@ -26,7 +26,8 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import org.dna.mqtt.moquette.messaging.spi.IMessaging;
import org.dna.mqtt.moquette.messaging.spi.IStorageService;
import org.dna.mqtt.moquette.messaging.spi.ISessionsStore;
import org.dna.mqtt.moquette.messaging.spi.IMessagesStore;
import org.dna.mqtt.moquette.messaging.spi.impl.events.*;
import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription;
import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.SubscriptionsStore;
@@ -56,7 +57,8 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
private RingBuffer<ValueEvent> m_ringBuffer;
private IStorageService m_storageService;
private IMessagesStore m_storageService;
private ISessionsStore m_sessionsStore;
private ExecutorService m_executor;
BatchEventProcessor<ValueEvent> m_eventProcessor;
@@ -201,16 +203,19 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
}
private void processInit(Properties props) {
m_storageService = new HawtDBStorageService();
HawtDBPersistentStore hawtStorage = new HawtDBPersistentStore();
m_storageService = hawtStorage;
m_storageService.initStore();
List<Subscription> storedSubscriptions = m_storageService.listAllSubscriptions();
m_sessionsStore = hawtStorage;
List<Subscription> storedSubscriptions = m_sessionsStore.listAllSubscriptions();
subscriptions.init(storedSubscriptions);
String passwdPath = props.getProperty("password_file");
String configPath = System.getProperty("moquette.path", null);
IAuthenticator authenticator = new FileAuthenticator(configPath, passwdPath);
m_processor.init(subscriptions, m_storageService, authenticator);
m_processor.init(subscriptions, m_storageService, m_sessionsStore, authenticator);
}
broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageServiceTest.java → broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBPersistentStoreTest.java
+3 -3
View File
@@ -29,16 +29,16 @@ import org.junit.Before;
*
* @author andrea
*/
public class HawtDBStorageServiceTest {
public class HawtDBPersistentStoreTest {
HawtDBStorageService m_storageService;
HawtDBPersistentStore m_storageService;
@Before
public void setUp() throws Exception {
File dbFile = new File(Server.STORAGE_FILE_PATH);
assertFalse(String.format("The DB storagefile %s already exists", Server.STORAGE_FILE_PATH), dbFile.exists());
m_storageService = new HawtDBStorageService();
m_storageService = new HawtDBPersistentStore();
m_storageService.initStore();
}
broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java
+13 -10
View File
@@ -23,7 +23,8 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.dna.mqtt.moquette.messaging.spi.IStorageService;
import org.dna.mqtt.moquette.messaging.spi.ISessionsStore;
import org.dna.mqtt.moquette.messaging.spi.IMessagesStore;
import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription;
import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.SubscriptionsStore;
@@ -58,7 +59,8 @@ public class ProtocolProcessorTest {
ConnectMessage connMsg;
ProtocolProcessor m_processor;
IStorageService m_storageService;
IMessagesStore m_storageService;
ISessionsStore m_sessionStore;
SubscriptionsStore subscriptions;
AbstractMessage m_receivedMessage;
MockAuthenticator m_mockAuthenticator;
@@ -144,8 +146,9 @@ public class ProtocolProcessorTest {
//sleep to let the messaging batch processor to process the initEvent
Thread.sleep(300);
m_storageService = new MemoryStorageService();
MemoryStorageService memStorage = new MemoryStorageService();
m_storageService = memStorage;
m_sessionStore = memStorage;
//m_storageService.initStore();
Map<String, String> users = new HashMap<String, String>();
@@ -155,7 +158,7 @@ public class ProtocolProcessorTest {
subscriptions = new SubscriptionsStore();
subscriptions.init(Collections.<Subscription>emptyList());
m_processor = new ProtocolProcessor();
m_processor.init(subscriptions, m_storageService, m_mockAuthenticator);
m_processor.init(subscriptions, m_storageService, m_sessionStore, m_mockAuthenticator);
}
@Test
@@ -263,7 +266,7 @@ public class ProtocolProcessorTest {
//simulate a connect that register a clientID to an IoSession
subs.init(Collections.<Subscription>emptyList());
m_processor.init(subs, m_storageService, null);
m_processor.init(subs, m_storageService, m_sessionStore, null);
ConnectMessage connectMessage = new ConnectMessage();
connectMessage.setProcotolVersion((byte)3);
connectMessage.setClientID(FAKE_CLIENT_ID);
@@ -302,7 +305,7 @@ public class ProtocolProcessorTest {
//simulate a connect that register a clientID to an IoSession
subs.init(Collections.<Subscription>emptyList());
m_processor.init(subs, m_storageService, null);
m_processor.init(subs, m_storageService, m_sessionStore, null);
MockReceiverChannel firstReceiverSession = new MockReceiverChannel();
ConnectMessage connectMessage = new ConnectMessage();
@@ -412,7 +415,7 @@ public class ProtocolProcessorTest {
subs.init(Collections.<Subscription>emptyList());
//simulate a connect that register a clientID to an IoSession
m_processor.init(subs, m_storageService, null);
m_processor.init(subs, m_storageService, m_sessionStore, null);
ConnectMessage connectMessage = new ConnectMessage();
connectMessage.setClientID(FAKE_PUBLISHER_ID);
connectMessage.setProcotolVersion((byte)3);
@@ -444,7 +447,7 @@ public class ProtocolProcessorTest {
List<Subscription> inactiveSubscriptions = Arrays.asList(inactiveSub);
when(mockedSubscriptions.matches(eq("/topic"))).thenReturn(inactiveSubscriptions);
m_processor = new ProtocolProcessor();
m_processor.init(mockedSubscriptions, m_storageService, null);
m_processor.init(mockedSubscriptions, m_storageService, m_sessionStore, null);
//Exercise
ByteBuffer buffer = ByteBuffer.allocate(5).put("Hello".getBytes());
@@ -464,7 +467,7 @@ public class ProtocolProcessorTest {
List<Subscription> inactiveSubscriptions = Arrays.asList(inactiveSub);
when(mockedSubscriptions.matches(eq("/topic"))).thenReturn(inactiveSubscriptions);
m_processor = new ProtocolProcessor();
m_processor.init(mockedSubscriptions, m_storageService, null);
m_processor.init(mockedSubscriptions, m_storageService, m_sessionStore, null);
//Exercise
ByteBuffer buffer = ByteBuffer.allocate(5).put("Hello".getBytes());