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