anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Removed unused persistence classes

Browse Source
This commit is contained in:
andsel
2018-11-14 22:13:42 +01:00
parent a36d926460
commit cfa478067e
15 changed files with 61 additions and 1384 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/interception/messages/InterceptAcknowledgedMessage.java
+59 -1
View File
@@ -16,10 +16,68 @@
package io.moquette.interception.messages;
import static io.moquette.spi.IMessagesStore.StoredMessage;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.handler.codec.mqtt.MqttQoS;
import java.io.Serializable;
public class InterceptAcknowledgedMessage implements InterceptMessage {
class StoredMessage implements Serializable {
private static final long serialVersionUID = 1755296138639817304L;
private MqttQoS m_qos;
final byte[] m_payload;
final String m_topic;
private boolean m_retained;
private String m_clientID;
public StoredMessage(byte[] message, MqttQoS qos, String topic) {
m_qos = qos;
m_payload = message;
m_topic = topic;
}
public void setQos(MqttQoS qos) {
this.m_qos = qos;
}
public MqttQoS getQos() {
return m_qos;
}
public String getTopic() {
return m_topic;
}
public String getClientID() {
return m_clientID;
}
public void setClientID(String m_clientID) {
this.m_clientID = m_clientID;
}
public ByteBuf getPayload() {
return Unpooled.copiedBuffer(m_payload);
}
public void setRetained(boolean retained) {
this.m_retained = retained;
}
public boolean isRetained() {
return m_retained;
}
@Override
public String toString() {
return "PublishEvent{clientID='" + m_clientID + '\'' + ", m_retain="
+ m_retained + ", m_qos=" + m_qos + ", m_topic='" + m_topic + '\'' + '}';
}
}
private final StoredMessage msg;
private final String username;
private final String topic;
broker/src/main/java/io/moquette/persistence/H2SubscriptionsRepository.java
+2 -2
View File
@@ -42,7 +42,7 @@ public class H2SubscriptionsRepository implements ISubscriptionsRepository {
}
@Override
public void removeSubscription(String topic, String clientID) {
public void removeSubscription(String topicFilter, String clientID) {
subscriptions.remove(topicFilter + "-" + clientID);
}
}
broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java
-69
View File
@@ -1,69 +0,0 @@
/*
* Copyright (c) 2012-2018 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 io.moquette.persistence;
import io.moquette.spi.IMatchingCondition;
import io.moquette.spi.IMessagesStore;
import io.moquette.broker.subscriptions.Topic;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.*;
public class MemoryMessagesStore implements IMessagesStore {
private static final Logger LOG = LoggerFactory.getLogger(MemoryMessagesStore.class);
private Map<Topic, StoredMessage> m_retainedStore = new HashMap<>();
MemoryMessagesStore() {
}
@Override
public void initStore() {
}
@Override
public void storeRetained(Topic topic, StoredMessage storedMessage) {
LOG.debug("Store retained message for topic={}, CId={}", topic, storedMessage.getClientID());
if (storedMessage.getClientID() == null) {
throw new IllegalArgumentException("Message to be persisted must have a not null client ID");
}
m_retainedStore.put(topic, storedMessage);
}
@Override
public Collection<StoredMessage> searchMatching(IMatchingCondition condition) {
LOG.debug("searchMatching scanning all retained messages, presents are {}", m_retainedStore.size());
List<StoredMessage> results = new ArrayList<>();
for (Map.Entry<Topic, StoredMessage> entry : m_retainedStore.entrySet()) {
StoredMessage storedMsg = entry.getValue();
if (condition.match(entry.getKey())) {
results.add(storedMsg);
}
}
return results;
}
@Override
public void cleanRetained(Topic topic) {
m_retainedStore.remove(topic);
}
}
broker/src/main/java/io/moquette/persistence/MemorySessionStore.java
-303
View File
@@ -1,303 +0,0 @@
/*
* Copyright (c) 2012-2018 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 io.moquette.persistence;
import io.moquette.spi.IMessagesStore.StoredMessage;
import io.moquette.spi.ISessionsStore;
import io.moquette.spi.ISubscriptionsStore;
import io.moquette.broker.subscriptions.Subscription;
import io.moquette.broker.subscriptions.Topic;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.*;
public class MemorySessionStore implements ISessionsStore, ISubscriptionsStore {
private static final Logger LOG = LoggerFactory.getLogger(MemorySessionStore.class);
static final int MAX_MESSAGE_QUEUE = 1024; // number of messages
static class Session {
final String clientID;
final Map<Topic, Subscription> subscriptions = new ConcurrentHashMap<>();
boolean cleanSession;
final BlockingQueue<StoredMessage> queue = new ArrayBlockingQueue<>(MAX_MESSAGE_QUEUE);
final Map<Integer, StoredMessage> secondPhaseStore = new ConcurrentHashMap<>();
final Map<Integer, StoredMessage> outboundFlightMessages =
Collections.synchronizedMap(new HashMap<Integer, StoredMessage>());
final Map<Integer, StoredMessage> inboundFlightMessages = new ConcurrentHashMap<>();
Session(String clientID, boolean cleanSession) {
this.clientID = clientID;
this.cleanSession = cleanSession;
}
}
private final Map<String, Session> sessions = new ConcurrentHashMap<>();
private ConcurrentNavigableMap<LocalDateTime, Set<String>> sessionsClosingTimes = new ConcurrentSkipListMap<>();
MemorySessionStore() {
}
private Session getSession(String clientID) {
Session session = sessions.get(clientID);
if (session == null) {
LOG.error("Can't find the session for client <{}>", clientID);
throw new RuntimeException("Can't find the session for client <" + clientID + ">");
}
return session;
}
@Override
public void removeSubscription(Topic topic, String clientID) {
LOG.debug("removeSubscription topic filter: {} for clientID: {}", topic, clientID);
getSession(clientID).subscriptions.remove(topic);
}
@Override
public void initStore() {
}
@Override
public ISubscriptionsStore subscriptionStore() {
return this;
}
@Override
public void addNewSubscription(Subscription newSubscription) {
final String clientID = newSubscription.getClientId();
Session session = sessions.get(clientID);
if (session == null) {
LOG.error("Can't find the session for client <{}>", clientID);
return;
}
session.subscriptions.put(newSubscription.getTopicFilter(), newSubscription);
}
@Override
public void wipeSubscriptions(String clientID) {
if (!sessions.containsKey(clientID)) {
LOG.error("Can't find the session for client <{}>", clientID);
return;
}
sessions.get(clientID).subscriptions.clear();
}
@Override
public boolean contains(String clientID) {
return sessions.containsKey(clientID);
}
@Override
public void createNewDurableSession(String clientID) {
Session innerSession = new Session(clientID, false);
sessions.put(clientID, innerSession);
}
@Override
public void removeDurableSession(String clientId) {
this.sessions.remove(clientId);
this.wipeSubscriptions(clientId);
}
@Override
public void updateCleanStatus(String clientId, boolean newCleanStatus) {
sessions.get(clientId).cleanSession = newCleanStatus;
}
@Override
public PersistentSession loadSessionByKey(String clientID) {
return new PersistentSession(clientID, sessions.get(clientID).cleanSession);
}
@Override
public Collection<PersistentSession> listAllSessions() {
Collection<PersistentSession> result = new ArrayList<>();
for (Session entry : sessions.values()) {
result.add(new PersistentSession(entry.clientID, entry.cleanSession));
}
return result;
}
@Override
public List<Subscription> listAllSubscriptions() {
List<Subscription> allSubscriptions = new ArrayList<>();
for (Session entry : sessions.values()) {
allSubscriptions.addAll(entry.subscriptions.values());
}
return allSubscriptions;
}
@Override
public Collection<Subscription> listClientSubscriptions(String clientID) {
final Session session = sessions.get(clientID);
if (session == null) {
throw new IllegalStateException("Asking for subscriptions of not persisted client: " + clientID);
}
return session.subscriptions.values();
}
@Override
public Subscription reload(Subscription subcription) {
String clientID = subcription.getClientId();
if (!sessions.containsKey(clientID)) {
LOG.error("Can't find the session for client <{}>", clientID);
return null;
}
Map<Topic, Subscription> subscriptions = sessions.get(clientID).subscriptions;
if (subscriptions == null || subscriptions.isEmpty()) {
return null;
}
return subscriptions.get(subcription.getTopicFilter());
}
@Override
public StoredMessage inFlightAck(String clientID, int messageID) {
return getSession(clientID).outboundFlightMessages.remove(messageID);
}
@Override
public void inFlight(String clientID, int messageID, StoredMessage msg) {
Session session = sessions.get(clientID);
if (session == null) {
LOG.error("Can't find the session for client <{}>", clientID);
return;
}
session.outboundFlightMessages.put(messageID, msg);
}
/**
* Return the next valid packetIdentifier for the given client session.
*/
@Override
public int nextPacketID(String clientID) {
if (!sessions.containsKey(clientID)) {
LOG.error("Can't find the session for client <{}>", clientID);
return -1;
}
Map<Integer, StoredMessage> m = sessions.get(clientID).outboundFlightMessages;
int maxId = m.keySet().isEmpty() ? 0 : Collections.max(m.keySet());
int nextPacketId = (maxId + 1) % 0xFFFF;
m.put(nextPacketId, null);
return nextPacketId;
}
@Override
public BlockingQueue<StoredMessage> queue(String clientID) {
if (!sessions.containsKey(clientID)) {
LOG.error("Can't find the session for client <{}>", clientID);
return null;
}
return sessions.get(clientID).queue;
}
@Override
public void dropQueue(String clientID) {
final Session session = sessions.get(clientID);
session.queue.clear();
session.outboundFlightMessages.clear();
session.inboundFlightMessages.clear();
}
@Override
public void moveInFlightToSecondPhaseAckWaiting(String clientID, int messageID, StoredMessage msg) {
LOG.info("Moving msg inflight second phase store, clientID <{}> messageID {}", clientID, messageID);
Session session = sessions.get(clientID);
if (session == null) {
LOG.error("Can't find the session for client <{}>", clientID);
return;
}
session.secondPhaseStore.put(messageID, msg);
session.outboundFlightMessages.put(messageID, msg);
}
@Override
public StoredMessage completeReleasedPublish(String clientID, int messageID) {
LOG.info("Acknowledged message in second phase, clientID <{}> messageID {}", clientID, messageID);
return getSession(clientID).secondPhaseStore.remove(messageID);
}
@Override
public int getInflightMessagesNo(String clientID) {
Session session = sessions.get(clientID);
if (session == null) {
LOG.error("Can't find the session for client <{}>", clientID);
return 0;
}
return session.inboundFlightMessages.size() +
// countPubReleaseWaitingPubComplete(clientID) +
session.outboundFlightMessages.size();
}
@Override
public int countPubReleaseWaitingPubComplete(String clientID) {
if (!sessions.containsKey(clientID)) {
LOG.error("Can't find the session for client <{}>", clientID);
return 0;
}
return sessions.get(clientID).secondPhaseStore.size();
}
@Override
public void removeTemporaryQoS2(String clientID) {
LOG.debug("Session cleanup for client <{}>", clientID);
Session session = sessions.get(clientID);
if (session == null) {
LOG.error("Can't find the session for client <{}>", clientID);
return;
}
LOG.info("Removing stored messages with QoS 2. ClientId={}", clientID);
session.secondPhaseStore.clear();
// TODO this missing last step breaks the junit test
//sessions.remove(clientID);
}
@Override
public synchronized void trackSessionClose(LocalDateTime when, String clientID) {
this.sessionsClosingTimes.putIfAbsent(when, new HashSet<>());
this.sessionsClosingTimes.computeIfPresent(when, (key, oldSet) -> {
oldSet.add(clientID);
return oldSet;
});
}
@Override
public Set<String> sessionOlderThan(LocalDateTime queryPin) {
final Set<String> results = new HashSet<>();
LocalDateTime keyBefore = this.sessionsClosingTimes.lowerKey(queryPin);
while (keyBefore != null) {
results.addAll(this.sessionsClosingTimes.get(keyBefore));
keyBefore = this.sessionsClosingTimes.lowerKey(keyBefore);
}
return results;
}
}
broker/src/main/java/io/moquette/persistence/MemoryStorageService.java
-58
View File
@@ -1,58 +0,0 @@
/*
* Copyright (c) 2012-2018 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 io.moquette.persistence;
import io.moquette.broker.config.IConfig;
import io.moquette.spi.IMessagesStore;
import io.moquette.spi.ISessionsStore;
import io.moquette.spi.IStore;
import java.util.concurrent.ScheduledExecutorService;
@Deprecated
public class MemoryStorageService implements IStore {
private MemorySessionStore m_sessionsStore;
private MemoryMessagesStore m_messagesStore;
// NB these params must be here "by contract" used in introspection instantiation used in
// ProtocolProcessorBootstrapper.instantiateConfiguredStore
public MemoryStorageService(IConfig props, ScheduledExecutorService scheduler) {
m_messagesStore = new MemoryMessagesStore();
m_sessionsStore = new MemorySessionStore();
m_messagesStore.initStore();
m_sessionsStore.initStore();
}
@Override
public IMessagesStore messagesStore() {
return m_messagesStore;
}
@Override
public ISessionsStore sessionsStore() {
return m_sessionsStore;
}
@Override
public void initStore() {
}
@Override
public void close() {
}
}
broker/src/main/java/io/moquette/persistence/PersistentSession.java
-34
View File
@@ -1,34 +0,0 @@
/*
* Copyright (c) 2012-2018 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 io.moquette.persistence;
import java.io.Serializable;
/**
* This is a DTO used to persist minimal status (clean session and activation status) of a
* session.
*/
public class PersistentSession implements Serializable {
private static final long serialVersionUID = 5052054783220481854L;
public final boolean cleanSession;
public final String clientID;
public PersistentSession(String clientID, boolean cleanSession) {
this.clientID = clientID;
this.cleanSession = cleanSession;
}
}
broker/src/main/java/io/moquette/spi/IMatchingCondition.java
-24
View File
@@ -1,24 +0,0 @@
/*
* Copyright (c) 2012-2018 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 io.moquette.spi;
import io.moquette.broker.subscriptions.Topic;
public interface IMatchingCondition {
boolean match(Topic key);
}
broker/src/main/java/io/moquette/spi/IMessagesStore.java
-111
View File
@@ -1,111 +0,0 @@
/*
* Copyright (c) 2012-2018 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 io.moquette.spi;
import io.moquette.broker.subscriptions.Topic;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.handler.codec.mqtt.MqttQoS;
import java.io.Serializable;
import java.util.Collection;
/**
* Defines the SPI to be implemented by a StorageService that handle persistence of messages
*/
public interface IMessagesStore {
class StoredMessage implements Serializable {
private static final long serialVersionUID = 1755296138639817304L;
private MqttQoS m_qos;
final byte[] m_payload;
final String m_topic;
private boolean m_retained;
private String m_clientID;
private MessageGUID m_guid;
public StoredMessage(byte[] message, MqttQoS qos, String topic) {
m_qos = qos;
m_payload = message;
m_topic = topic;
}
public void setQos(MqttQoS qos) {
this.m_qos = qos;
}
public MqttQoS getQos() {
return m_qos;
}
public String getTopic() {
return m_topic;
}
public void setGuid(MessageGUID guid) {
this.m_guid = guid;
}
public MessageGUID getGuid() {
return m_guid;
}
public String getClientID() {
return m_clientID;
}
public void setClientID(String m_clientID) {
this.m_clientID = m_clientID;
}
public ByteBuf getPayload() {
return Unpooled.copiedBuffer(m_payload);
}
public void setRetained(boolean retained) {
this.m_retained = retained;
}
public boolean isRetained() {
return m_retained;
}
@Override
public String toString() {
return "PublishEvent{clientID='" + m_clientID + '\'' + ", m_retain="
+ m_retained + ", m_qos=" + m_qos + ", m_topic='" + m_topic + '\'' + '}';
}
}
/**
* Used to initialize all persistent store structures
*/
void initStore();
/**
* Return a list of retained messages that satisfy the condition.
*
* @param condition
* the condition to match during the search.
* @return the collection of matching messages.
*/
Collection<StoredMessage> searchMatching(IMatchingCondition condition);
void cleanRetained(Topic topic);
void storeRetained(Topic topic, StoredMessage storedMessage);
}
broker/src/main/java/io/moquette/spi/ISessionsStore.java
-131
View File
@@ -1,131 +0,0 @@
/*
* Copyright (c) 2012-2018 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 io.moquette.spi;
import io.moquette.persistence.PersistentSession;
import io.moquette.spi.IMessagesStore.StoredMessage;
import java.time.LocalDateTime;
import java.util.Collection;
import java.util.Queue;
import java.util.Set;
/**
* Store used to handle the persistence of the subscriptions tree.
*/
@Deprecated
public interface ISessionsStore {
void initStore();
ISubscriptionsStore subscriptionStore();
/**
* @param clientID
* the session client ID.
* @return true iff there is a session stored for the client
*/
boolean contains(String clientID);
void createNewDurableSession(String clientID);
void removeDurableSession(String clientId);
void updateCleanStatus(String clientID, boolean cleanSession);
PersistentSession loadSessionByKey(String clientID);
/**
* Returns all the sessions
*
* @return the collection of all stored client sessions.
*/
Collection<PersistentSession> listAllSessions();
StoredMessage inFlightAck(String clientID, int messageID);
/**
* Save the message msg with messageID, clientID as in flight
*
* @param clientID
* the client ID
* @param messageID
* the message ID
* @param msg
* the message to put in flight zone
*/
void inFlight(String clientID, int messageID, StoredMessage msg);
/**
* Return the next valid packetIdentifier for the given client session.
*
* @param clientID
* the clientID requesting next packet id.
* @return the next valid id.
*/
int nextPacketID(String clientID);
/**
* List the published retained messages for the session
*
* @param clientID
* the client ID owning the queue.
* @return the queue of messages.
*/
Queue<StoredMessage> queue(String clientID);
void dropQueue(String clientID);
void moveInFlightToSecondPhaseAckWaiting(String clientID, int messageID, StoredMessage msg);
/**
* @param clientID
* the client ID accessing the second phase.
* @param messageID
* the message ID that reached the second phase.
* @return the guid of message just acked.
*/
StoredMessage completeReleasedPublish(String clientID, int messageID);
/**
* Returns the number of inflight messages for the given client ID
*
* @param clientID target client.
* @return count of pending in flight publish messages.
*/
int getInflightMessagesNo(String clientID);
/**
* Returns the number of second-phase ACK pending messages for the given client ID
*
* @param clientID target client.
* @return count of pending in flight publish messages.
*/
int countPubReleaseWaitingPubComplete(String clientID);
void removeTemporaryQoS2(String clientID);
/**
* Tracks the closing time of a persistent session
* */
void trackSessionClose(LocalDateTime when, String clientID);
/**
* List the sessions ids closed before the pin date.
* */
Set<String> sessionOlderThan(LocalDateTime queryPin);
}
broker/src/main/java/io/moquette/spi/IStore.java
-39
View File
@@ -1,39 +0,0 @@
/*
* Copyright (c) 2012-2018 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 io.moquette.spi;
@Deprecated
public interface IStore {
void initStore();
void close();
/**
* Factory method to instantiate the messages store.
*
* @return the create instance of IMessagesStore.
* */
IMessagesStore messagesStore();
/**
* Factory method to instantiate the session store.
*
* @return the create instance of ISessionsStore.
* */
ISessionsStore sessionsStore();
}
broker/src/main/java/io/moquette/spi/ISubscriptionsStore.java
-69
View File
@@ -1,69 +0,0 @@
/*
* Copyright (c) 2012-2018 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 io.moquette.spi;
import io.moquette.broker.subscriptions.Subscription;
import io.moquette.broker.subscriptions.Topic;
import java.util.Collection;
import java.util.List;
/**
* @deprecated look at ISubscriptionsRepository
*/
@Deprecated
public interface ISubscriptionsStore {
/**
* Add a new subscription to the session
*
* @param newSubscription the subscription to add.
*/
void addNewSubscription(Subscription newSubscription);
/**
* Removed a specific subscription
*
* @param topic the topic of the subscription.
* @param clientID the session client.
*/
void removeSubscription(Topic topic, String clientID);
/**
* Remove all the subscriptions of the session
*
* @param clientID the client ID
*/
void wipeSubscriptions(String clientID);
/**
* Return all topic filters to recreate the subscription tree.
*/
List<Subscription> listAllSubscriptions();
/**
* Load from storage all the subscriptions of the specified client.
*/
Collection<Subscription> listClientSubscriptions(String clientID);
/**
* @param subcription the subscription to reaload from storage.
* @return the subscription stored by clientID and topicFilter, if any else null;
*/
Subscription reload(Subscription subcription);
}
broker/src/main/java/io/moquette/spi/MessageGUID.java
-54
View File
@@ -1,54 +0,0 @@
/*
* Copyright (c) 2012-2018 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 io.moquette.spi;
import java.io.Serializable;
/**
* Value object for GUIDs of messages.
*/
public class MessageGUID implements Serializable {
private static final long serialVersionUID = 4315161987111542406L;
private final String guid;
public MessageGUID(String guid) {
this.guid = guid;
}
@Override
public boolean equals(Object o) {
if (this == o)
return true;
if (o == null || getClass() != o.getClass())
return false;
MessageGUID that = (MessageGUID) o;
return guid != null ? guid.equals(that.guid) : that.guid == null;
}
@Override
public int hashCode() {
return guid != null ? guid.hashCode() : 0;
}
@Override
public String toString() {
return "MessageGUID{" + "guid='" + guid + '\'' + '}';
}
}
h2_storage/src/main/java/io/moquette/persistence/h2/H2MessagesStore.java
-78
View File
@@ -1,78 +0,0 @@
/*
* Copyright (c) 2012-2018 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 io.moquette.persistence.h2;
import io.moquette.spi.IMatchingCondition;
import io.moquette.spi.IMessagesStore;
import io.moquette.broker.subscriptions.Topic;
import org.h2.mvstore.Cursor;
import org.h2.mvstore.MVMap;
import org.h2.mvstore.MVStore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.*;
class H2MessagesStore implements IMessagesStore {
private static final Logger LOG = LoggerFactory.getLogger(H2MessagesStore.class);
private final MVStore mvStore;
private MVMap<Topic, StoredMessage> retainedStore;
public H2MessagesStore(MVStore mvStore) {
this.mvStore = mvStore;
}
@Override
public void initStore() {
retainedStore = mvStore.openMap("retained");
LOG.info("Initialized message H2 store");
}
@Override
public void storeRetained(Topic topic, StoredMessage storedMessage) {
LOG.debug("Store retained message for topic={}, CId={}", topic, storedMessage.getClientID());
if (storedMessage.getClientID() == null) {
throw new IllegalArgumentException("Message to be persisted must have a not null client ID");
}
retainedStore.put(topic, storedMessage);
}
@Override
public Collection<StoredMessage> searchMatching(IMatchingCondition condition) {
LOG.debug("Scanning retained messages");
List<StoredMessage> results = new ArrayList<>();
Cursor<Topic, StoredMessage> mapCursor = retainedStore.cursor(null);
while (mapCursor.hasNext()) {
Topic topic = mapCursor.next();
if (topic != null && condition.match(topic))
results.add(mapCursor.getValue());
}
LOG.trace("Retained messages have been scanned matchingMessages={}", results);
return results;
}
@Override
public void cleanRetained(Topic topic) {
LOG.debug("Cleaning retained messages. Topic={}", topic);
retainedStore.remove(topic);
}
}
h2_storage/src/main/java/io/moquette/persistence/h2/H2PersistentStore.java
-90
View File
@@ -1,90 +0,0 @@
/*
* Copyright (c) 2012-2018 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 io.moquette.persistence.h2;
import io.moquette.BrokerConstants;
import io.moquette.broker.config.IConfig;
import io.moquette.spi.IMessagesStore;
import io.moquette.spi.ISessionsStore;
import io.moquette.spi.IStore;
import org.h2.mvstore.MVStore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
public class H2PersistentStore implements IStore {
private static final Logger LOG = LoggerFactory.getLogger(H2PersistentStore.class);
private final String storePath;
private final int autosaveInterval; // in seconds
protected final ScheduledExecutorService scheduler;
private MVStore mvStore;
private IMessagesStore messageStore;
private ISessionsStore sessionsStore;
public H2PersistentStore(IConfig props, ScheduledExecutorService scheduler) {
this.storePath = props.getProperty(BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME, "");
final String autosaveProp = props.getProperty(BrokerConstants.AUTOSAVE_INTERVAL_PROPERTY_NAME, "30");
this.autosaveInterval = Integer.parseInt(autosaveProp);
this.scheduler = scheduler;
}
@SuppressWarnings("FutureReturnValueIgnored")
@Override
public void initStore() {
LOG.info("Initializing H2 store");
if (storePath == null || storePath.isEmpty()) {
LOG.warn("H2 store file path is empty, using in-memory store");
mvStore = MVStore.open(null);
} else {
mvStore = new MVStore.Builder()
.fileName(storePath)
.autoCommitDisabled()
.open();
}
LOG.info("Scheduling H2 commit task");
scheduler.scheduleWithFixedDelay(() -> {
LOG.debug("Committing to H2");
mvStore.commit();
}, this.autosaveInterval, this.autosaveInterval, TimeUnit.SECONDS);
messageStore = new H2MessagesStore(mvStore);
messageStore.initStore();
sessionsStore = new H2SessionsStore(mvStore);
sessionsStore.initStore();
}
@Override
public void close() {
}
@Override
public IMessagesStore messagesStore() {
return messageStore;
}
@Override
public ISessionsStore sessionsStore() {
return sessionsStore;
}
}
h2_storage/src/main/java/io/moquette/persistence/h2/H2SessionsStore.java
-321
View File
@@ -1,321 +0,0 @@
/*
* Copyright (c) 2012-2018 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 io.moquette.persistence.h2;
import io.moquette.persistence.PersistentSession;
import io.moquette.spi.IMessagesStore;
import io.moquette.spi.IMessagesStore.StoredMessage;
import io.moquette.spi.ISessionsStore;
import io.moquette.spi.ISubscriptionsStore;
import io.moquette.broker.subscriptions.Subscription;
import io.moquette.broker.subscriptions.Topic;
import org.h2.mvstore.MVMap;
import org.h2.mvstore.MVStore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
public class H2SessionsStore implements ISessionsStore, ISubscriptionsStore {
private static final Logger LOG = LoggerFactory.getLogger(H2SessionsStore.class);
private final MVStore mvStore;
private ConcurrentMap<String, PersistentSession> sessions;
// maps clientID->[MessageId -> msg]
private ConcurrentMap<String, ConcurrentMap<Integer, StoredMessage>> outboundFlightMessages;
// map clientID <-> set of currently in flight packet identifiers
private Map<String, Set<Integer>> inFlightIds;
// maps clientID->[MessageId -> guid]
private ConcurrentMap<String, ConcurrentMap<Integer, StoredMessage>> secondPhaseStore;
private MVMap<LocalDateTime, Set<String>> sessionsClosingTimes;
H2SessionsStore(MVStore mvStore) {
this.mvStore = mvStore;
}
@Override
public void initStore() {
this.sessions = mvStore.openMap("sessions");
this.outboundFlightMessages = mvStore.openMap("outboundFlight");
this.inFlightIds = mvStore.openMap("inflightPacketIDs");
this.secondPhaseStore = mvStore.openMap("secondPhase");
this.sessionsClosingTimes = mvStore.openMap("sessionsCreationTimes");
LOG.info("Initialized sessions H2 store");
}
@Override
public ISubscriptionsStore subscriptionStore() {
return this;
}
//TODO move in the subscription directory
@Override
public void addNewSubscription(Subscription newSubscription) {
LOG.info("Adding new subscription CId={}, topics={}", newSubscription.getClientId(),
newSubscription.getTopicFilter());
final String clientID = newSubscription.getClientId();
final MVMap<Object, Object> sessionSubscriptions = this.mvStore.openMap(subscriptionsMapName(clientID));
sessionSubscriptions.put(newSubscription.getTopicFilter(), newSubscription);
if (LOG.isTraceEnabled()) {
LOG.trace("Subscription has been added CId={}, topics={}, clientSubscriptions={}",
newSubscription.getClientId(), newSubscription.getTopicFilter(), sessionSubscriptions);
}
}
//TODO move in the subscription directory
@Override
public void removeSubscription(Topic topicFilter, String clientID) {
LOG.info("Removing subscription. CId={}, topics={}", clientID, topicFilter);
if (!this.mvStore.hasMap(subscriptionsMapName(clientID))) {
return;
}
final MVMap<Object, Object> sessionsSubscriptions = this.mvStore.openMap(subscriptionsMapName(clientID));
sessionsSubscriptions.remove(topicFilter);
if (LOG.isDebugEnabled()) {
LOG.debug("Subscription has been removed. CId={}, topics={}, clientSubscriptions={}", clientID,
topicFilter, sessionsSubscriptions);
}
}
@Override
public void wipeSubscriptions(String sessionID) {
LOG.info("Wiping subscriptions. CId={}", sessionID);
if (!this.mvStore.hasMap(subscriptionsMapName(sessionID))) {
return;
}
final MVMap<Object, Object> subscriptions = this.mvStore.openMap(subscriptionsMapName(sessionID));
this.mvStore.removeMap(subscriptions);
if (LOG.isDebugEnabled()) {
LOG.debug("Subscriptions have been removed. CId={}, clientSubscriptions={}", sessionID, subscriptions);
}
}
@Override
public List<Subscription> listAllSubscriptions() {
LOG.debug("Retrieving existing subscriptions...");
List<Subscription> subscriptions = new ArrayList<>();
for (String clientID : this.sessions.keySet()) {
final String mapName = subscriptionsMapName(clientID);
ConcurrentMap<Topic, Subscription> clientSubscriptions = this.mvStore.openMap(mapName);
subscriptions.addAll(clientSubscriptions.values());
}
LOG.debug("Existing subscriptions has been retrieved Result={}", subscriptions);
return subscriptions;
}
@Override
public Collection<Subscription> listClientSubscriptions(String clientID) {
ConcurrentMap<Topic, Subscription> clientSubscriptions = this.mvStore.openMap(subscriptionsMapName(clientID));
if (clientSubscriptions == null) {
throw new IllegalStateException("Asking for subscriptions of not persisted client: " + clientID);
}
return clientSubscriptions.values();
}
@Override
public Subscription reload(Subscription subscription) {
final String subscriptionsMapName = subscriptionsMapName(subscription.getClientId());
ConcurrentMap<Topic, Subscription> clientSubscriptions = this.mvStore.openMap(subscriptionsMapName);
LOG.debug("Retrieving subscriptions CId={}, subscriptions={}", subscription.getClientId(), clientSubscriptions);
return clientSubscriptions.get(subscription.getTopicFilter());
}
private String subscriptionsMapName(String clientId) {
return "subscriptions_" + clientId;
}
@Override
public boolean contains(String clientID) {
return this.sessions.containsKey(clientID);
}
@Override
public void createNewDurableSession(String clientID) {
this.sessions.putIfAbsent(clientID, new PersistentSession(clientID, false));
}
@Override
public void removeDurableSession(String clientId) {
this.sessions.remove(clientId);
this.wipeSubscriptions(clientId);
}
@Override
public void updateCleanStatus(String clientId, boolean newCleanStatus) {
PersistentSession updatedSession = new PersistentSession(clientId, newCleanStatus);
this.sessions.put(clientId, updatedSession);
}
@Override
public PersistentSession loadSessionByKey(String clientID) {
return this.sessions.get(clientID);
}
@Override
public Collection<PersistentSession> listAllSessions() {
return this.sessions.values();
}
@Override
public StoredMessage inFlightAck(String clientID, int messageID) {
LOG.debug("Acknowledging inflight message CId={}, messageId={}", clientID, messageID);
ConcurrentMap<Integer, StoredMessage> m = this.outboundFlightMessages.get(clientID);
if (m == null) {
LOG.error("Can't find the inFlight record for client <{}>", clientID);
throw new RuntimeException("Can't find the inFlight record for client <" + clientID + ">");
}
StoredMessage msg = m.remove(messageID);
this.outboundFlightMessages.put(clientID, m);
// remove from the ids store
Set<Integer> inFlightForClient = this.inFlightIds.get(clientID);
if (inFlightForClient != null) {
inFlightForClient.remove(messageID);
}
return msg;
}
@Override
public void inFlight(String clientID, int messageID, StoredMessage msg) {
ConcurrentMap<Integer, IMessagesStore.StoredMessage> messages = this.outboundFlightMessages.get(clientID);
if (messages == null) {
messages = new ConcurrentHashMap<>();
}
messages.put(messageID, msg);
this.outboundFlightMessages.put(clientID, messages);
}
@Override
public int nextPacketID(String clientID) {
LOG.debug("Generating next packet ID CId={}", clientID);
Set<Integer> inFlightForClient = this.inFlightIds.get(clientID);
if (inFlightForClient == null) {
int nextPacketId = 1;
inFlightForClient = Collections.newSetFromMap(new ConcurrentHashMap<>());
inFlightForClient.add(nextPacketId);
this.inFlightIds.put(clientID, inFlightForClient);
return nextPacketId;
}
int maxId = inFlightForClient.isEmpty() ? 0 : Collections.max(inFlightForClient);
int nextPacketId = (maxId % 0xFFFF) + 1;
inFlightForClient.add(nextPacketId);
LOG.debug("Next packet ID has been generated CId={}, result={}", clientID, nextPacketId);
return nextPacketId;
}
@Override
public Queue<StoredMessage> queue(String clientID) {
return new H2PersistentQueue<>(this.mvStore, clientID);
}
@Override
public void dropQueue(String clientID) {
H2PersistentQueue.dropQueue(this.mvStore, clientID);
this.outboundFlightMessages.remove(clientID);
this.inFlightIds.remove(clientID);
}
@Override
public void moveInFlightToSecondPhaseAckWaiting(String clientID, int messageID, StoredMessage msg) {
LOG.debug("Moving inflight message to 2nd phase ack state CId={}, messageID={}", clientID, messageID);
ConcurrentMap<Integer, StoredMessage> m = this.secondPhaseStore.get(clientID);
if (m == null) {
String error = String.format("Can't find the inFlight record for client <%s> during the second phase of " +
"QoS2 pub", clientID);
LOG.error(error);
throw new RuntimeException(error);
}
m.put(messageID, msg);
this.outboundFlightMessages.put(clientID, m);
}
@Override
public StoredMessage completeReleasedPublish(String clientID, int messageID) {
LOG.debug("Processing second phase ACK CId={}, messageId={}", clientID, messageID);
final ConcurrentMap<Integer, StoredMessage> m = this.secondPhaseStore.get(clientID);
if (m == null) {
String error = String.format("Can't find the inFlight record for client <%s> during the second phase " +
"acking of QoS2 pub", clientID);
LOG.error(error);
throw new RuntimeException(error);
}
StoredMessage msg = m.remove(messageID);
this.secondPhaseStore.put(clientID, m);
return msg;
}
@Override
public int getInflightMessagesNo(String clientID) {
int totalInflight = 0;
ConcurrentMap<Integer, StoredMessage> inflightPerClient = this.mvStore.openMap(inboundStoreForClient(clientID));
if (inflightPerClient != null) {
totalInflight += inflightPerClient.size();
}
// totalInflight += countPubReleaseWaitingPubComplete(clientID);
Map<Integer, StoredMessage> outboundPerClient = outboundFlightMessages.get(clientID);
if (outboundPerClient != null) {
totalInflight += outboundPerClient.size();
}
return totalInflight;
}
private static String inboundStoreForClient(String clientID) {
return "inboundInflight_" + clientID;
}
@Override
public int countPubReleaseWaitingPubComplete(String clientID) {
if (!this.secondPhaseStore.containsKey(clientID))
return 0;
return this.secondPhaseStore.get(clientID).size();
}
@Override
public void removeTemporaryQoS2(String clientID) {
LOG.info("Removing stored messages with QoS 2. ClientId={}", clientID);
this.secondPhaseStore.remove(clientID);
}
@Override
public synchronized void trackSessionClose(LocalDateTime when, String clientID) {
this.sessionsClosingTimes.putIfAbsent(when, new HashSet<>());
this.sessionsClosingTimes.computeIfPresent(when, (key, oldSet) -> {
oldSet.add(clientID);
return oldSet;
});
}
@Override
public Set<String> sessionOlderThan(LocalDateTime queryPin) {
final Set<String> results = new HashSet<>();
LocalDateTime keyBefore = this.sessionsClosingTimes.lowerKey(queryPin);
while (keyBefore != null) {
results.addAll(this.sessionsClosingTimes.get(keyBefore));
keyBefore = this.sessionsClosingTimes.lowerKey(keyBefore);
}
return results;
}
}