From f3e6b71e09937f97b61b2de2fe9423ceb47e3811 Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Mon, 1 Sep 2014 22:06:27 +0200 Subject: [PATCH] Dropped HawtDB persistence --- broker/pom.xml | 63 +--- .../spi/impl/HawtDBPersistentStore.java | 300 ------------------ .../spi/impl/MemoryStorageService.java | 21 +- .../messaging/spi/impl/ProtocolProcessor.java | 4 +- .../messaging/spi/impl/SimpleMessaging.java | 3 - .../spi/impl/ProtocolProcessorTest.java | 2 +- .../subscriptions/SubscriptionsStoreTest.java | 21 -- bundle/pom.xml | 21 +- distribution/pom.xml | 14 +- 9 files changed, 25 insertions(+), 424 deletions(-) delete mode 100644 broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBPersistentStore.java diff --git a/broker/pom.xml b/broker/pom.xml index 924f28d8..52c30264 100644 --- a/broker/pom.xml +++ b/broker/pom.xml @@ -50,8 +50,9 @@ 1.7.5 runtime - - + + + - + io.netty @@ -106,18 +107,6 @@ 4.0.8.Final - - org.fusesource.hawtbuf - hawtbuf - 1.9 - - - - org.fusesource.hawtdb - hawtdb - 1.6 - - org.mapdb mapdb @@ -181,46 +170,6 @@ - - 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 deleted file mode 100644 index 20183e3d..00000000 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBPersistentStore.java +++ /dev/null @@ -1,300 +0,0 @@ -/* - * 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.impl; - -import java.io.File; -import java.io.IOException; -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.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; -import org.dna.mqtt.moquette.proto.messages.AbstractMessage; -import org.dna.mqtt.moquette.server.Server; -import org.fusesource.hawtbuf.codec.StringCodec; -import org.fusesource.hawtdb.api.*; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - - -/** - * Implementation of IMessagesStore and ISessionsStore backed by HawtDB - */ -public class HawtDBPersistentStore implements IMessagesStore, ISessionsStore { - - private static final Logger LOG = LoggerFactory.getLogger(HawtDBPersistentStore.class); - - private MultiIndexFactory m_multiIndexFactory; - private PageFileFactory pageFactory; - - //maps clientID to the list of pending messages stored - //TODO move in a multimap because only Qos1 and QoS2 are stored here and they have messageID(key of secondary map) - private SortedIndex> m_persistentMessageStore; - private SortedIndex m_retainedStore; - //bind clientID+MsgID -> evt message published - private SortedIndex m_inflightStore; - //bind clientID+MsgID -> evt message published - private SortedIndex m_qos2Store; - - //persistent Map of clientID, set of Subscriptions - private SortedIndex> m_persistentSubscriptions; - - public HawtDBPersistentStore() { - String storeFile = Server.STORAGE_FILE_PATH; - - pageFactory = new PageFileFactory(); - File tmpFile; - try { - tmpFile = new File(storeFile); - tmpFile.createNewFile(); - } catch (IOException ex) { - LOG.error(null, ex); - throw new MQTTException("Can't create temp file for subscriptions storage [" + storeFile + "]", ex); - } - pageFactory.setFile(tmpFile); - pageFactory.open(); - PageFile pageFile = pageFactory.getPageFile(); - m_multiIndexFactory = new MultiIndexFactory(pageFile); - } - - - public void initStore() { - initRetainedStore(); - //init the message store for QoS 1/2 messages in clean sessions - initPersistentMessageStore(); - initInflightMessageStore(); - initPersistentSubscriptions(); - initPersistentQoS2MessageStore(); - } - - private void initRetainedStore() { - BTreeIndexFactory indexFactory = new BTreeIndexFactory(); - indexFactory.setKeyCodec(StringCodec.INSTANCE); - - m_retainedStore = (SortedIndex) m_multiIndexFactory.openOrCreate("retained", indexFactory); - } - - - private void initPersistentMessageStore() { - BTreeIndexFactory> indexFactory = new BTreeIndexFactory>(); - indexFactory.setKeyCodec(StringCodec.INSTANCE); - - m_persistentMessageStore = (SortedIndex>) m_multiIndexFactory.openOrCreate("persistedMessages", indexFactory); - } - - private void initPersistentSubscriptions() { - BTreeIndexFactory> indexFactory = new BTreeIndexFactory>(); - indexFactory.setKeyCodec(StringCodec.INSTANCE); - - m_persistentSubscriptions = (SortedIndex>) m_multiIndexFactory.openOrCreate("subscriptions", indexFactory); - } - - /** - * Initialize the message store used to handle the temporary storage of QoS 1,2 - * messages in flight. - */ - private void initInflightMessageStore() { - BTreeIndexFactory indexFactory = new BTreeIndexFactory(); - indexFactory.setKeyCodec(StringCodec.INSTANCE); - - m_inflightStore = (SortedIndex) m_multiIndexFactory.openOrCreate("inflight", indexFactory); - } - - private void initPersistentQoS2MessageStore() { - BTreeIndexFactory indexFactory = new BTreeIndexFactory(); - indexFactory.setKeyCodec(StringCodec.INSTANCE); - - m_qos2Store = (SortedIndex) m_multiIndexFactory.openOrCreate("qos2Store", indexFactory); - } - - @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)); - } - } - - @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) { - 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); - //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); - } - - 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) { - StoredPublishEvent storedEvt = convertToStored(evt); - m_inflightStore.put(publishKey, storedEvt); - } - - 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); - } - } - - public void wipeSubscriptions(String clientID) { - m_persistentSubscriptions.remove(clientID); - } - - public List listAllSubscriptions() { - List allSubscriptions = new ArrayList(); - for (Map.Entry> entry : m_persistentSubscriptions) { - 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"); - try { - pageFactory.close(); - } catch (IOException ex) { - LOG.error(null, ex); - } - } - - /*-------- 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/messaging/spi/impl/MemoryStorageService.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java index cabc1b2c..17dbf554 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java @@ -16,19 +16,14 @@ package org.dna.mqtt.moquette.messaging.spi.impl; import java.nio.ByteBuffer; -import java.util.ArrayList; +import java.util.*; + import org.dna.mqtt.moquette.messaging.spi.IMatchingCondition; 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; -import java.util.Collection; -import java.util.HashMap; -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; @@ -38,7 +33,7 @@ import org.slf4j.LoggerFactory; public class MemoryStorageService implements IMessagesStore, ISessionsStore { private Map> m_persistentSubscriptions = new HashMap>(); - private Map m_retainedStore = new HashMap(); + private Map m_retainedStore = new HashMap(); //TODO move in a multimap because only Qos1 and QoS2 are stored here and they have messageID(key of secondary map) private Map> m_persistentMessageStore = new HashMap>(); private Map m_inflightStore = new HashMap(); @@ -63,18 +58,18 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore { //store the message to the topic byte[] raw = new byte[message.remaining()]; message.get(raw); - m_retainedStore.put(topic, new HawtDBPersistentStore.StoredMessage(raw, qos, topic)); + m_retainedStore.put(topic, new StoredMessage(raw, qos, topic)); } } @Override - public Collection searchMatching(IMatchingCondition condition) { + public Collection searchMatching(IMatchingCondition condition) { LOG.debug("searchMatching scanning all retained messages, presents are {}", m_retainedStore.size()); - List results = new ArrayList(); + List results = new ArrayList(); - for (Map.Entry entry : m_retainedStore.entrySet()) { - HawtDBPersistentStore.StoredMessage storedMsg = entry.getValue(); + for (Map.Entry entry : m_retainedStore.entrySet()) { + StoredMessage storedMsg = entry.getValue(); if (condition.match(entry.getKey())) { results.add(storedMsg); } diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java index 452ab740..abeebe5f 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java @@ -603,13 +603,13 @@ class ProtocolProcessor implements EventHandler { subscriptions.add(newSubscription); //scans retained messages to be published to the new subscription - Collection messages = m_messagesStore.searchMatching(new IMatchingCondition() { + Collection messages = m_messagesStore.searchMatching(new IMatchingCondition() { public boolean match(String key) { return SubscriptionsStore.matchTopics(key, topic); } }); - for (HawtDBPersistentStore.StoredMessage storedMsg : messages) { + for (IMessagesStore.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); 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 6eb70c9e..7d322bcc 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 @@ -20,7 +20,6 @@ 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; @@ -34,9 +33,7 @@ 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; import org.dna.mqtt.moquette.server.IAuthenticator; import org.dna.mqtt.moquette.server.ServerChannel; import org.slf4j.Logger; diff --git a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java index a4f5324d..79db84e1 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java @@ -570,7 +570,7 @@ public class ProtocolProcessorTest { m_processor.processPublish(m_session, cleanPubMsg); //Verify - Collection messages = m_storageService.searchMatching(new IMatchingCondition() { + Collection messages = m_storageService.searchMatching(new IMatchingCondition() { public boolean match(String key) { return SubscriptionsStore.matchTopics(key, FAKE_TOPIC); } diff --git a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStoreTest.java b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStoreTest.java index 71201e21..26e1d9d6 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStoreTest.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStoreTest.java @@ -22,11 +22,7 @@ import java.util.Arrays; import java.util.Collections; import java.util.List; -import org.dna.mqtt.moquette.messaging.spi.impl.MemoryStorageService; import org.dna.mqtt.moquette.proto.messages.AbstractMessage; -import org.fusesource.hawtdb.api.MultiIndexFactory; -import org.fusesource.hawtdb.api.PageFile; -import org.fusesource.hawtdb.api.PageFileFactory; import org.junit.After; import static org.junit.Assert.*; import org.junit.Before; @@ -38,7 +34,6 @@ import org.junit.Test; */ public class SubscriptionsStoreTest { - private File hawtFile; private SubscriptionsStore store; public SubscriptionsStoreTest() { @@ -46,26 +41,10 @@ public class SubscriptionsStoreTest { @Before public void setUp() throws IOException { - hawtFile = File.createTempFile("moquette_persistent_store", ".dat"); - store = new SubscriptionsStore(); -// store.init(hawtFile.getAbsolutePath()); - PageFileFactory pageFactory = new PageFileFactory(); - - pageFactory.setFile(hawtFile); - pageFactory.open(); - PageFile pageFile = pageFactory.getPageFile(); - MultiIndexFactory multiIndexFactory = new MultiIndexFactory(pageFile); store.init(Collections.emptyList()); } - @After - public void tearDown() { - if (hawtFile.exists()) { - hawtFile.delete(); - } - } - @Test public void testSplitTopic() throws ParseException { List tokens = store.splitTopic("finance/stock/ibm"); diff --git a/bundle/pom.xml b/bundle/pom.xml index 6bf83367..7a3f17ef 100644 --- a/bundle/pom.xml +++ b/bundle/pom.xml @@ -72,11 +72,11 @@ - + io.netty @@ -119,21 +119,14 @@ 4.0.8.Final provided - + - org.fusesource.hawtbuf - hawtbuf - 1.9 + org.mapdb + mapdb + 1.0.6 provided - - - org.fusesource.hawtdb - hawtdb - 1.6 - provided - - + junit junit diff --git a/distribution/pom.xml b/distribution/pom.xml index bd106a79..f21eda60 100644 --- a/distribution/pom.xml +++ b/distribution/pom.xml @@ -32,19 +32,7 @@ moquette-netty-parser ${project.version} - - - org.fusesource.hawtbuf - hawtbuf - 1.9 - - - - org.fusesource.hawtdb - hawtdb - 1.6 - - + org.slf4j slf4j-api