Switched from HawtDB to MapDP store
This commit is contained in:
selva.andre
@@ -117,6 +117,12 @@
|
||||
<artifactId>hawtdb</artifactId>
|
||||
<version>1.6</version>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.mapdb</groupId>
|
||||
<artifactId>mapdb</artifactId>
|
||||
<version>1.0.6</version>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>junit</groupId>
|
||||
|
||||
@@ -15,10 +15,10 @@
|
||||
*/
|
||||
package org.dna.mqtt.moquette.messaging.spi;
|
||||
|
||||
import java.io.Serializable;
|
||||
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.HawtDBPersistentStore.StoredMessage;
|
||||
|
||||
import java.util.Collection;
|
||||
import java.util.List;
|
||||
@@ -28,6 +28,30 @@ import java.util.List;
|
||||
*/
|
||||
public interface IMessagesStore {
|
||||
|
||||
public static class StoredMessage implements Serializable {
|
||||
final AbstractMessage.QOSType m_qos;
|
||||
final byte[] m_payload;
|
||||
final String m_topic;
|
||||
|
||||
public StoredMessage(byte[] message, AbstractMessage.QOSType qos, String topic) {
|
||||
m_qos = qos;
|
||||
m_payload = message;
|
||||
m_topic = topic;
|
||||
}
|
||||
|
||||
public AbstractMessage.QOSType getQos() {
|
||||
return m_qos;
|
||||
}
|
||||
|
||||
public ByteBuffer getPayload() {
|
||||
return (ByteBuffer) ByteBuffer.allocate(m_payload.length).put(m_payload).flip();
|
||||
}
|
||||
|
||||
public String getTopic() {
|
||||
return m_topic;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Used to initialize all persistent store structures
|
||||
* */
|
||||
|
||||
+3
-27
@@ -17,7 +17,6 @@ package org.dna.mqtt.moquette.messaging.spi.impl;
|
||||
|
||||
import java.io.File;
|
||||
import java.io.IOException;
|
||||
import java.io.Serializable;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.util.*;
|
||||
import org.dna.mqtt.moquette.MQTTException;
|
||||
@@ -40,31 +39,6 @@ import org.slf4j.LoggerFactory;
|
||||
*/
|
||||
public class HawtDBPersistentStore implements IMessagesStore, ISessionsStore {
|
||||
|
||||
public static class StoredMessage implements Serializable {
|
||||
AbstractMessage.QOSType m_qos;
|
||||
byte[] m_payload;
|
||||
String m_topic;
|
||||
|
||||
StoredMessage(byte[] message, AbstractMessage.QOSType qos, String topic) {
|
||||
m_qos = qos;
|
||||
m_payload = message;
|
||||
m_topic = topic;
|
||||
}
|
||||
|
||||
AbstractMessage.QOSType getQos() {
|
||||
return m_qos;
|
||||
}
|
||||
|
||||
ByteBuffer getPayload() {
|
||||
return (ByteBuffer) ByteBuffer.allocate(m_payload.length).put(m_payload).flip();
|
||||
}
|
||||
|
||||
String getTopic() {
|
||||
return m_topic;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(HawtDBPersistentStore.class);
|
||||
|
||||
private MultiIndexFactory m_multiIndexFactory;
|
||||
@@ -184,6 +158,7 @@ public class HawtDBPersistentStore implements IMessagesStore, ISessionsStore {
|
||||
return results;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void storePublishForFuture(PublishEvent evt) {
|
||||
List<StoredPublishEvent> storedEvents;
|
||||
String clientID = evt.getClientID();
|
||||
@@ -210,7 +185,8 @@ public class HawtDBPersistentStore implements IMessagesStore, ISessionsStore {
|
||||
}
|
||||
return liveEvts;
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
public void cleanPersistedPublishMessage(String clientID, int messageID) {
|
||||
List<StoredPublishEvent> events = m_persistentMessageStore.get(clientID);
|
||||
if (events == null) {
|
||||
|
||||
@@ -19,6 +19,8 @@ import com.lmax.disruptor.BatchEventProcessor;
|
||||
import com.lmax.disruptor.EventHandler;
|
||||
import com.lmax.disruptor.RingBuffer;
|
||||
import com.lmax.disruptor.SequenceBarrier;
|
||||
|
||||
import java.io.File;
|
||||
import java.util.List;
|
||||
import java.util.Properties;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
@@ -31,6 +33,7 @@ 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;
|
||||
import org.dna.mqtt.moquette.messaging.spi.persistence.MapDBPersistentStore;
|
||||
import org.dna.mqtt.moquette.proto.messages.PubCompMessage;
|
||||
import org.dna.mqtt.moquette.proto.messages.*;
|
||||
import org.dna.mqtt.moquette.server.Constants;
|
||||
@@ -155,10 +158,12 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
|
||||
}
|
||||
|
||||
private void processInit(Properties props) {
|
||||
HawtDBPersistentStore hawtStorage = new HawtDBPersistentStore();
|
||||
m_storageService = hawtStorage;
|
||||
//TODO use a property to select the storage path
|
||||
MapDBPersistentStore mapStorage = new MapDBPersistentStore();
|
||||
m_storageService = mapStorage;
|
||||
m_sessionsStore = mapStorage;
|
||||
|
||||
m_storageService.initStore();
|
||||
m_sessionsStore = hawtStorage;
|
||||
|
||||
List<Subscription> storedSubscriptions = m_sessionsStore.listAllSubscriptions();
|
||||
subscriptions.init(storedSubscriptions);
|
||||
|
||||
+255
@@ -0,0 +1,255 @@
|
||||
/*
|
||||
* Copyright (c) 2012-2014 The original author or authors
|
||||
* ------------------------------------------------------
|
||||
* All rights reserved. This program and the accompanying materials
|
||||
* are made available under the terms of the Eclipse Public License v1.0
|
||||
* and Apache License v2.0 which accompanies this distribution.
|
||||
*
|
||||
* The Eclipse Public License is available at
|
||||
* http://www.eclipse.org/legal/epl-v10.html
|
||||
*
|
||||
* The Apache License v2.0 is available at
|
||||
* http://www.opensource.org/licenses/apache2.0.php
|
||||
*
|
||||
* You may elect to redistribute this code under either of these licenses.
|
||||
*/
|
||||
|
||||
package org.dna.mqtt.moquette.messaging.spi.persistence;
|
||||
|
||||
import org.dna.mqtt.moquette.MQTTException;
|
||||
import org.dna.mqtt.moquette.messaging.spi.IMatchingCondition;
|
||||
import org.dna.mqtt.moquette.messaging.spi.IMessagesStore;
|
||||
import org.dna.mqtt.moquette.messaging.spi.ISessionsStore;
|
||||
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;
|
||||
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
|
||||
import static org.dna.mqtt.moquette.server.Server.STORAGE_FILE_PATH;
|
||||
import org.mapdb.DB;
|
||||
import org.mapdb.DBMaker;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.io.File;
|
||||
import java.io.IOException;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.util.*;
|
||||
import java.util.concurrent.ConcurrentMap;
|
||||
|
||||
/**
|
||||
* MapDB main persistence implementation
|
||||
*/
|
||||
public class MapDBPersistentStore implements IMessagesStore, ISessionsStore {
|
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(MapDBPersistentStore.class);
|
||||
|
||||
private ConcurrentMap<String, StoredMessage> m_retainedStore;
|
||||
//maps clientID to the list of pending messages stored
|
||||
private ConcurrentMap<String, List<StoredPublishEvent>> m_persistentMessageStore;
|
||||
//bind clientID+MsgID -> evt message published
|
||||
private ConcurrentMap<String, StoredPublishEvent> m_inflightStore;
|
||||
//bind clientID+MsgID -> evt message published
|
||||
private ConcurrentMap<String, StoredPublishEvent> m_qos2Store;
|
||||
//persistent Map of clientID, set of Subscriptions
|
||||
private ConcurrentMap<String, Set<Subscription>> m_persistentSubscriptions;
|
||||
private DB m_db;
|
||||
|
||||
@Override
|
||||
public void initStore() {
|
||||
File tmpFile;
|
||||
try {
|
||||
tmpFile = new File(STORAGE_FILE_PATH);
|
||||
tmpFile.createNewFile();
|
||||
} catch (IOException ex) {
|
||||
LOG.error(null, ex);
|
||||
throw new MQTTException("Can't create temp file for subscriptions storage [" + STORAGE_FILE_PATH + "]", ex);
|
||||
}
|
||||
m_db = DBMaker.newFileDB(tmpFile)
|
||||
.closeOnJvmShutdown()
|
||||
.make();
|
||||
m_retainedStore = m_db.getHashMap("retained");
|
||||
m_persistentMessageStore = m_db.getHashMap("persistedMessages");
|
||||
m_inflightStore = m_db.getHashMap("inflight");
|
||||
m_persistentSubscriptions = m_db.getHashMap("subscriptions");
|
||||
m_qos2Store = m_db.getHashMap("qos2Store");
|
||||
}
|
||||
|
||||
@Override
|
||||
public void cleanRetained(String topic) {
|
||||
m_retainedStore.remove(topic);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void storeRetained(String topic, ByteBuffer message, AbstractMessage.QOSType qos) {
|
||||
if (!message.hasRemaining()) {
|
||||
//clean the message from topic
|
||||
m_retainedStore.remove(topic);
|
||||
} else {
|
||||
//store the message to the topic
|
||||
byte[] raw = new byte[message.remaining()];
|
||||
message.get(raw);
|
||||
m_retainedStore.put(topic, new StoredMessage(raw, qos, topic));
|
||||
}
|
||||
m_db.commit();
|
||||
}
|
||||
|
||||
@Override
|
||||
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.entrySet()) {
|
||||
StoredMessage storedMsg = entry.getValue();
|
||||
if (condition.match(entry.getKey())) {
|
||||
results.add(storedMsg);
|
||||
}
|
||||
}
|
||||
|
||||
return results;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void storePublishForFuture(PublishEvent evt) {
|
||||
List<StoredPublishEvent> storedEvents;
|
||||
String clientID = evt.getClientID();
|
||||
if (!m_persistentMessageStore.containsKey(clientID)) {
|
||||
storedEvents = new ArrayList<StoredPublishEvent>();
|
||||
} else {
|
||||
storedEvents = m_persistentMessageStore.get(clientID);
|
||||
}
|
||||
storedEvents.add(convertToStored(evt));
|
||||
m_persistentMessageStore.put(clientID, storedEvents);
|
||||
m_db.commit();
|
||||
//NB rewind the evt message content
|
||||
LOG.debug("Stored published message for client <{}> on topic <{}>", clientID, evt.getTopic());
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<PublishEvent> retrievePersistedPublishes(String clientID) {
|
||||
List<StoredPublishEvent> storedEvts = m_persistentMessageStore.get(clientID);
|
||||
if (storedEvts == null) {
|
||||
return null;
|
||||
}
|
||||
List<PublishEvent> liveEvts = new ArrayList<PublishEvent>();
|
||||
for (StoredPublishEvent storedEvt : storedEvts) {
|
||||
liveEvts.add(convertFromStored(storedEvt));
|
||||
}
|
||||
return liveEvts;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void cleanPersistedPublishMessage(String clientID, int messageID) {
|
||||
List<StoredPublishEvent> events = m_persistentMessageStore.get(clientID);
|
||||
if (events == null) {
|
||||
return;
|
||||
}
|
||||
StoredPublishEvent toRemoveEvt = null;
|
||||
for (StoredPublishEvent evt : events) {
|
||||
if (evt.getMessageID() == messageID) {
|
||||
toRemoveEvt = evt;
|
||||
}
|
||||
}
|
||||
events.remove(toRemoveEvt);
|
||||
m_persistentMessageStore.put(clientID, events);
|
||||
m_db.commit();
|
||||
}
|
||||
|
||||
public void cleanPersistedPublishes(String clientID) {
|
||||
m_persistentMessageStore.remove(clientID);
|
||||
m_db.commit();
|
||||
}
|
||||
|
||||
public void cleanInFlight(String msgID) {
|
||||
m_inflightStore.remove(msgID);
|
||||
m_db.commit();
|
||||
}
|
||||
|
||||
public void addInFlight(PublishEvent evt, String publishKey) {
|
||||
StoredPublishEvent storedEvt = convertToStored(evt);
|
||||
m_inflightStore.put(publishKey, storedEvt);
|
||||
m_db.commit();
|
||||
}
|
||||
|
||||
public void addNewSubscription(Subscription newSubscription, String clientID) {
|
||||
LOG.debug("addNewSubscription invoked with subscription {} for client {}", newSubscription, clientID);
|
||||
if (!m_persistentSubscriptions.containsKey(clientID)) {
|
||||
LOG.debug("clientID {} is a newcome, creating it's subscriptions set", clientID);
|
||||
m_persistentSubscriptions.put(clientID, new HashSet<Subscription>());
|
||||
}
|
||||
|
||||
Set<Subscription> subs = m_persistentSubscriptions.get(clientID);
|
||||
if (!subs.contains(newSubscription)) {
|
||||
LOG.debug("updating clientID {} subscriptions set with new subscription", clientID);
|
||||
//TODO check the subs doesn't contain another subscription to the same topic with different
|
||||
Subscription existingSubscription = null;
|
||||
for (Subscription scanSub : subs) {
|
||||
if (newSubscription.getTopicFilter().equals(scanSub.getTopicFilter())) {
|
||||
existingSubscription = scanSub;
|
||||
break;
|
||||
}
|
||||
}
|
||||
if (existingSubscription != null) {
|
||||
subs.remove(existingSubscription);
|
||||
}
|
||||
subs.add(newSubscription);
|
||||
m_persistentSubscriptions.put(clientID, subs);
|
||||
LOG.debug("clientID {} subscriptions set now is {}", clientID, subs);
|
||||
}
|
||||
m_db.commit();
|
||||
}
|
||||
|
||||
public void wipeSubscriptions(String clientID) {
|
||||
m_persistentSubscriptions.remove(clientID);
|
||||
m_db.commit();
|
||||
}
|
||||
|
||||
public List<Subscription> listAllSubscriptions() {
|
||||
List<Subscription> allSubscriptions = new ArrayList<Subscription>();
|
||||
for (Map.Entry<String, Set<Subscription>> entry : m_persistentSubscriptions.entrySet()) {
|
||||
allSubscriptions.addAll(entry.getValue());
|
||||
}
|
||||
LOG.debug("retrieveAllSubscriptions returning subs {}", allSubscriptions);
|
||||
return allSubscriptions;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean contains(String clientID) {
|
||||
return m_persistentSubscriptions.containsKey(clientID);
|
||||
}
|
||||
|
||||
public void close() {
|
||||
LOG.debug("closing disk storage");
|
||||
this.m_db.commit();
|
||||
this.m_db.close();
|
||||
}
|
||||
|
||||
/*-------- QoS 2 storage management --------------*/
|
||||
public void persistQoS2Message(String publishKey, PublishEvent evt) {
|
||||
LOG.debug("persistQoS2Message store pubKey: {}, evt: {}", publishKey, evt);
|
||||
m_qos2Store.put(publishKey, convertToStored(evt));
|
||||
}
|
||||
|
||||
public void removeQoS2Message(String publishKey) {
|
||||
m_qos2Store.remove(publishKey);
|
||||
}
|
||||
|
||||
public PublishEvent retrieveQoS2Message(String publishKey) {
|
||||
StoredPublishEvent storedEvt = m_qos2Store.get(publishKey);
|
||||
return convertFromStored(storedEvt);
|
||||
}
|
||||
|
||||
private StoredPublishEvent convertToStored(PublishEvent evt) {
|
||||
StoredPublishEvent storedEvt = new StoredPublishEvent(evt);
|
||||
return storedEvt;
|
||||
}
|
||||
|
||||
private PublishEvent convertFromStored(StoredPublishEvent evt) {
|
||||
byte[] message = evt.getMessage();
|
||||
ByteBuffer bbmessage = ByteBuffer.wrap(message);
|
||||
//bbmessage.flip();
|
||||
PublishEvent liveEvt = new PublishEvent(evt.getTopic(), evt.getQos(),
|
||||
bbmessage, evt.isRetain(), evt.getClientID(), evt.getMessageID());
|
||||
return liveEvt;
|
||||
}
|
||||
}
|
||||
@@ -31,8 +31,8 @@ public class Server {
|
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(Server.class);
|
||||
|
||||
public static final String STORAGE_FILE_PATH = System.getProperty("user.home") +
|
||||
File.separator + "moquette_store.hawtdb";
|
||||
public static final String STORAGE_FILE_PATH = System.getProperty("user.home") +
|
||||
File.separator + "moquette_store.mapdb";
|
||||
|
||||
private ServerAcceptor m_acceptor;
|
||||
SimpleMessaging messaging;
|
||||
|
||||
+4
-3
@@ -18,6 +18,7 @@ package org.dna.mqtt.moquette.messaging.spi.impl;
|
||||
import java.io.File;
|
||||
import java.util.List;
|
||||
import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription;
|
||||
import org.dna.mqtt.moquette.messaging.spi.persistence.MapDBPersistentStore;
|
||||
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
|
||||
import org.dna.mqtt.moquette.server.Server;
|
||||
import org.junit.After;
|
||||
@@ -29,16 +30,16 @@ import org.junit.Before;
|
||||
*
|
||||
* @author andrea
|
||||
*/
|
||||
public class HawtDBPersistentStoreTest {
|
||||
public class MapDBPersistentStoreTest {
|
||||
|
||||
HawtDBPersistentStore m_storageService;
|
||||
MapDBPersistentStore 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 HawtDBPersistentStore();
|
||||
m_storageService = new MapDBPersistentStore();
|
||||
m_storageService.initStore();
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user