diff --git a/broker/pom.xml b/broker/pom.xml index 945d4a03..924f28d8 100644 --- a/broker/pom.xml +++ b/broker/pom.xml @@ -117,6 +117,12 @@ hawtdb 1.6 + + + org.mapdb + mapdb + 1.0.6 + junit diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessagesStore.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessagesStore.java index d21b9918..459dcdb6 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessagesStore.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessagesStore.java @@ -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 * */ diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBPersistentStore.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBPersistentStore.java index e53aad8f..20183e3d 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBPersistentStore.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBPersistentStore.java @@ -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 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 events = m_persistentMessageStore.get(clientID); if (events == null) { diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java index bbd4567a..6eb70c9e 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java @@ -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 { } 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 storedSubscriptions = m_sessionsStore.listAllSubscriptions(); subscriptions.init(storedSubscriptions); diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/persistence/MapDBPersistentStore.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/persistence/MapDBPersistentStore.java new file mode 100644 index 00000000..ab768e9d --- /dev/null +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/persistence/MapDBPersistentStore.java @@ -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 m_retainedStore; + //maps clientID to the list of pending messages stored + private ConcurrentMap> m_persistentMessageStore; + //bind clientID+MsgID -> evt message published + private ConcurrentMap m_inflightStore; + //bind clientID+MsgID -> evt message published + private ConcurrentMap m_qos2Store; + //persistent Map of clientID, set of Subscriptions + private ConcurrentMap> 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 searchMatching(IMatchingCondition condition) { + LOG.debug("searchMatching scanning all retained messages, presents are {}", m_retainedStore.size()); + + List results = new ArrayList(); + + for (Map.Entry 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 storedEvents; + String clientID = evt.getClientID(); + if (!m_persistentMessageStore.containsKey(clientID)) { + storedEvents = new ArrayList(); + } 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 retrievePersistedPublishes(String clientID) { + List storedEvts = m_persistentMessageStore.get(clientID); + if (storedEvts == null) { + return null; + } + List liveEvts = new ArrayList(); + for (StoredPublishEvent storedEvt : storedEvts) { + liveEvts.add(convertFromStored(storedEvt)); + } + return liveEvts; + } + + @Override + public void cleanPersistedPublishMessage(String clientID, int messageID) { + List 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()); + } + + Set 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 listAllSubscriptions() { + List allSubscriptions = new ArrayList(); + for (Map.Entry> 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; + } +} diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/Server.java b/broker/src/main/java/org/dna/mqtt/moquette/server/Server.java index e57aed3a..c1398d46 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/server/Server.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/server/Server.java @@ -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; diff --git a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBPersistentStoreTest.java b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/MapDBPersistentStoreTest.java similarity index 91% rename from broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBPersistentStoreTest.java rename to broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/MapDBPersistentStoreTest.java index e1bf0648..ef15006d 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBPersistentStoreTest.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/MapDBPersistentStoreTest.java @@ -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(); }