Dropped HawtDB persistence
This commit is contained in:
selva.andre
-300
@@ -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<String, List<StoredPublishEvent>> m_persistentMessageStore;
|
||||
private SortedIndex<String, StoredMessage> m_retainedStore;
|
||||
//bind clientID+MsgID -> evt message published
|
||||
private SortedIndex<String, StoredPublishEvent> m_inflightStore;
|
||||
//bind clientID+MsgID -> evt message published
|
||||
private SortedIndex<String, StoredPublishEvent> m_qos2Store;
|
||||
|
||||
//persistent Map of clientID, set of Subscriptions
|
||||
private SortedIndex<String, Set<Subscription>> 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<String, StoredMessage> indexFactory = new BTreeIndexFactory<String, StoredMessage>();
|
||||
indexFactory.setKeyCodec(StringCodec.INSTANCE);
|
||||
|
||||
m_retainedStore = (SortedIndex<String, StoredMessage>) m_multiIndexFactory.openOrCreate("retained", indexFactory);
|
||||
}
|
||||
|
||||
|
||||
private void initPersistentMessageStore() {
|
||||
BTreeIndexFactory<String, List<StoredPublishEvent>> indexFactory = new BTreeIndexFactory<String, List<StoredPublishEvent>>();
|
||||
indexFactory.setKeyCodec(StringCodec.INSTANCE);
|
||||
|
||||
m_persistentMessageStore = (SortedIndex<String, List<StoredPublishEvent>>) m_multiIndexFactory.openOrCreate("persistedMessages", indexFactory);
|
||||
}
|
||||
|
||||
private void initPersistentSubscriptions() {
|
||||
BTreeIndexFactory<String, Set<Subscription>> indexFactory = new BTreeIndexFactory<String, Set<Subscription>>();
|
||||
indexFactory.setKeyCodec(StringCodec.INSTANCE);
|
||||
|
||||
m_persistentSubscriptions = (SortedIndex<String, Set<Subscription>>) 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<String, StoredPublishEvent> indexFactory = new BTreeIndexFactory<String, StoredPublishEvent>();
|
||||
indexFactory.setKeyCodec(StringCodec.INSTANCE);
|
||||
|
||||
m_inflightStore = (SortedIndex<String, StoredPublishEvent>) m_multiIndexFactory.openOrCreate("inflight", indexFactory);
|
||||
}
|
||||
|
||||
private void initPersistentQoS2MessageStore() {
|
||||
BTreeIndexFactory<String, StoredPublishEvent> indexFactory = new BTreeIndexFactory<String, StoredPublishEvent>();
|
||||
indexFactory.setKeyCodec(StringCodec.INSTANCE);
|
||||
|
||||
m_qos2Store = (SortedIndex<String, StoredPublishEvent>) 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<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) {
|
||||
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);
|
||||
//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);
|
||||
}
|
||||
|
||||
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<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);
|
||||
}
|
||||
}
|
||||
|
||||
public void wipeSubscriptions(String clientID) {
|
||||
m_persistentSubscriptions.remove(clientID);
|
||||
}
|
||||
|
||||
public List<Subscription> listAllSubscriptions() {
|
||||
List<Subscription> allSubscriptions = new ArrayList<Subscription>();
|
||||
for (Map.Entry<String, Set<Subscription>> 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;
|
||||
}
|
||||
}
|
||||
+8
-13
@@ -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<String, Set<Subscription>> m_persistentSubscriptions = new HashMap<String, Set<Subscription>>();
|
||||
private Map<String, HawtDBPersistentStore.StoredMessage> m_retainedStore = new HashMap<String, HawtDBPersistentStore.StoredMessage>();
|
||||
private Map<String, StoredMessage> m_retainedStore = new HashMap<String, 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>();
|
||||
@@ -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<HawtDBPersistentStore.StoredMessage> searchMatching(IMatchingCondition condition) {
|
||||
public Collection<StoredMessage> searchMatching(IMatchingCondition condition) {
|
||||
LOG.debug("searchMatching scanning all retained messages, presents are {}", m_retainedStore.size());
|
||||
|
||||
List<HawtDBPersistentStore.StoredMessage> results = new ArrayList<HawtDBPersistentStore.StoredMessage>();
|
||||
List<StoredMessage> results = new ArrayList<StoredMessage>();
|
||||
|
||||
for (Map.Entry<String, HawtDBPersistentStore.StoredMessage> entry : m_retainedStore.entrySet()) {
|
||||
HawtDBPersistentStore.StoredMessage storedMsg = entry.getValue();
|
||||
for (Map.Entry<String, StoredMessage> entry : m_retainedStore.entrySet()) {
|
||||
StoredMessage storedMsg = entry.getValue();
|
||||
if (condition.match(entry.getKey())) {
|
||||
results.add(storedMsg);
|
||||
}
|
||||
|
||||
+2
-2
@@ -603,13 +603,13 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
|
||||
subscriptions.add(newSubscription);
|
||||
|
||||
//scans retained messages to be published to the new subscription
|
||||
Collection<HawtDBPersistentStore.StoredMessage> messages = m_messagesStore.searchMatching(new IMatchingCondition() {
|
||||
Collection<IMessagesStore.StoredMessage> 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);
|
||||
|
||||
@@ -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;
|
||||
|
||||
Reference in New Issue
Block a user