From cfa478067eb8ae370bfa9ebcb89d219bb50b2e56 Mon Sep 17 00:00:00 2001 From: andsel Date: Wed, 14 Nov 2018 22:13:42 +0100 Subject: [PATCH] Removed unused persistence classes --- .../InterceptAcknowledgedMessage.java | 60 +++- .../H2SubscriptionsRepository.java | 4 +- .../persistence/MemoryMessagesStore.java | 69 ---- .../persistence/MemorySessionStore.java | 303 ----------------- .../persistence/MemoryStorageService.java | 58 ---- .../persistence/PersistentSession.java | 34 -- .../io/moquette/spi/IMatchingCondition.java | 24 -- .../java/io/moquette/spi/IMessagesStore.java | 111 ------ .../java/io/moquette/spi/ISessionsStore.java | 131 ------- .../src/main/java/io/moquette/spi/IStore.java | 39 --- .../io/moquette/spi/ISubscriptionsStore.java | 69 ---- .../java/io/moquette/spi/MessageGUID.java | 54 --- .../persistence/h2/H2MessagesStore.java | 78 ----- .../persistence/h2/H2PersistentStore.java | 90 ----- .../persistence/h2/H2SessionsStore.java | 321 ------------------ 15 files changed, 61 insertions(+), 1384 deletions(-) delete mode 100644 broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java delete mode 100644 broker/src/main/java/io/moquette/persistence/MemorySessionStore.java delete mode 100644 broker/src/main/java/io/moquette/persistence/MemoryStorageService.java delete mode 100644 broker/src/main/java/io/moquette/persistence/PersistentSession.java delete mode 100644 broker/src/main/java/io/moquette/spi/IMatchingCondition.java delete mode 100644 broker/src/main/java/io/moquette/spi/IMessagesStore.java delete mode 100644 broker/src/main/java/io/moquette/spi/ISessionsStore.java delete mode 100644 broker/src/main/java/io/moquette/spi/IStore.java delete mode 100644 broker/src/main/java/io/moquette/spi/ISubscriptionsStore.java delete mode 100644 broker/src/main/java/io/moquette/spi/MessageGUID.java delete mode 100644 h2_storage/src/main/java/io/moquette/persistence/h2/H2MessagesStore.java delete mode 100644 h2_storage/src/main/java/io/moquette/persistence/h2/H2PersistentStore.java delete mode 100644 h2_storage/src/main/java/io/moquette/persistence/h2/H2SessionsStore.java diff --git a/broker/src/main/java/io/moquette/interception/messages/InterceptAcknowledgedMessage.java b/broker/src/main/java/io/moquette/interception/messages/InterceptAcknowledgedMessage.java index a74d3f8e..d33ede4f 100644 --- a/broker/src/main/java/io/moquette/interception/messages/InterceptAcknowledgedMessage.java +++ b/broker/src/main/java/io/moquette/interception/messages/InterceptAcknowledgedMessage.java @@ -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; diff --git a/broker/src/main/java/io/moquette/persistence/H2SubscriptionsRepository.java b/broker/src/main/java/io/moquette/persistence/H2SubscriptionsRepository.java index e733ebfc..b4d5ba41 100644 --- a/broker/src/main/java/io/moquette/persistence/H2SubscriptionsRepository.java +++ b/broker/src/main/java/io/moquette/persistence/H2SubscriptionsRepository.java @@ -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); } } diff --git a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java deleted file mode 100644 index ab0daec5..00000000 --- a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java +++ /dev/null @@ -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 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 searchMatching(IMatchingCondition condition) { - LOG.debug("searchMatching scanning all retained messages, presents are {}", m_retainedStore.size()); - - List results = new ArrayList<>(); - - for (Map.Entry entry : m_retainedStore.entrySet()) { - StoredMessage storedMsg = entry.getValue(); - if (condition.match(entry.getKey())) { - results.add(storedMsg); - } - } - - return results; - } - - @Override - public void cleanRetained(Topic topic) { - m_retainedStore.remove(topic); - } -} diff --git a/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java b/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java deleted file mode 100644 index a8b238e8..00000000 --- a/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java +++ /dev/null @@ -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 subscriptions = new ConcurrentHashMap<>(); - boolean cleanSession; - final BlockingQueue queue = new ArrayBlockingQueue<>(MAX_MESSAGE_QUEUE); - final Map secondPhaseStore = new ConcurrentHashMap<>(); - final Map outboundFlightMessages = - Collections.synchronizedMap(new HashMap()); - final Map inboundFlightMessages = new ConcurrentHashMap<>(); - - Session(String clientID, boolean cleanSession) { - this.clientID = clientID; - this.cleanSession = cleanSession; - } - } - - private final Map sessions = new ConcurrentHashMap<>(); - private ConcurrentNavigableMap> 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 listAllSessions() { - Collection result = new ArrayList<>(); - for (Session entry : sessions.values()) { - result.add(new PersistentSession(entry.clientID, entry.cleanSession)); - } - return result; - } - - @Override - public List listAllSubscriptions() { - List allSubscriptions = new ArrayList<>(); - for (Session entry : sessions.values()) { - allSubscriptions.addAll(entry.subscriptions.values()); - } - return allSubscriptions; - } - - @Override - public Collection 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 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 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 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 sessionOlderThan(LocalDateTime queryPin) { - final Set 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; - } -} diff --git a/broker/src/main/java/io/moquette/persistence/MemoryStorageService.java b/broker/src/main/java/io/moquette/persistence/MemoryStorageService.java deleted file mode 100644 index 3aea9de4..00000000 --- a/broker/src/main/java/io/moquette/persistence/MemoryStorageService.java +++ /dev/null @@ -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() { - } -} diff --git a/broker/src/main/java/io/moquette/persistence/PersistentSession.java b/broker/src/main/java/io/moquette/persistence/PersistentSession.java deleted file mode 100644 index 6728f3f0..00000000 --- a/broker/src/main/java/io/moquette/persistence/PersistentSession.java +++ /dev/null @@ -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; - } -} diff --git a/broker/src/main/java/io/moquette/spi/IMatchingCondition.java b/broker/src/main/java/io/moquette/spi/IMatchingCondition.java deleted file mode 100644 index 47bb8c41..00000000 --- a/broker/src/main/java/io/moquette/spi/IMatchingCondition.java +++ /dev/null @@ -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); -} diff --git a/broker/src/main/java/io/moquette/spi/IMessagesStore.java b/broker/src/main/java/io/moquette/spi/IMessagesStore.java deleted file mode 100644 index d90048e8..00000000 --- a/broker/src/main/java/io/moquette/spi/IMessagesStore.java +++ /dev/null @@ -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 searchMatching(IMatchingCondition condition); - - void cleanRetained(Topic topic); - - void storeRetained(Topic topic, StoredMessage storedMessage); -} diff --git a/broker/src/main/java/io/moquette/spi/ISessionsStore.java b/broker/src/main/java/io/moquette/spi/ISessionsStore.java deleted file mode 100644 index 8960f44c..00000000 --- a/broker/src/main/java/io/moquette/spi/ISessionsStore.java +++ /dev/null @@ -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 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 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 sessionOlderThan(LocalDateTime queryPin); -} diff --git a/broker/src/main/java/io/moquette/spi/IStore.java b/broker/src/main/java/io/moquette/spi/IStore.java deleted file mode 100644 index 8af6e71c..00000000 --- a/broker/src/main/java/io/moquette/spi/IStore.java +++ /dev/null @@ -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(); -} diff --git a/broker/src/main/java/io/moquette/spi/ISubscriptionsStore.java b/broker/src/main/java/io/moquette/spi/ISubscriptionsStore.java deleted file mode 100644 index 0c96ce97..00000000 --- a/broker/src/main/java/io/moquette/spi/ISubscriptionsStore.java +++ /dev/null @@ -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 listAllSubscriptions(); - - /** - * Load from storage all the subscriptions of the specified client. - */ - Collection 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); -} diff --git a/broker/src/main/java/io/moquette/spi/MessageGUID.java b/broker/src/main/java/io/moquette/spi/MessageGUID.java deleted file mode 100644 index d4b43655..00000000 --- a/broker/src/main/java/io/moquette/spi/MessageGUID.java +++ /dev/null @@ -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 + '\'' + '}'; - } -} diff --git a/h2_storage/src/main/java/io/moquette/persistence/h2/H2MessagesStore.java b/h2_storage/src/main/java/io/moquette/persistence/h2/H2MessagesStore.java deleted file mode 100644 index d282a433..00000000 --- a/h2_storage/src/main/java/io/moquette/persistence/h2/H2MessagesStore.java +++ /dev/null @@ -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 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 searchMatching(IMatchingCondition condition) { - LOG.debug("Scanning retained messages"); - List results = new ArrayList<>(); - - Cursor 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); - } -} diff --git a/h2_storage/src/main/java/io/moquette/persistence/h2/H2PersistentStore.java b/h2_storage/src/main/java/io/moquette/persistence/h2/H2PersistentStore.java deleted file mode 100644 index 6d14bbac..00000000 --- a/h2_storage/src/main/java/io/moquette/persistence/h2/H2PersistentStore.java +++ /dev/null @@ -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; - } - -} diff --git a/h2_storage/src/main/java/io/moquette/persistence/h2/H2SessionsStore.java b/h2_storage/src/main/java/io/moquette/persistence/h2/H2SessionsStore.java deleted file mode 100644 index dda7ed07..00000000 --- a/h2_storage/src/main/java/io/moquette/persistence/h2/H2SessionsStore.java +++ /dev/null @@ -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 sessions; - // maps clientID->[MessageId -> msg] - private ConcurrentMap> outboundFlightMessages; - // map clientID <-> set of currently in flight packet identifiers - private Map> inFlightIds; - // maps clientID->[MessageId -> guid] - private ConcurrentMap> secondPhaseStore; - private MVMap> 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 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 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 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 listAllSubscriptions() { - LOG.debug("Retrieving existing subscriptions..."); - List subscriptions = new ArrayList<>(); - for (String clientID : this.sessions.keySet()) { - final String mapName = subscriptionsMapName(clientID); - ConcurrentMap clientSubscriptions = this.mvStore.openMap(mapName); - subscriptions.addAll(clientSubscriptions.values()); - } - LOG.debug("Existing subscriptions has been retrieved Result={}", subscriptions); - return subscriptions; - } - - @Override - public Collection listClientSubscriptions(String clientID) { - ConcurrentMap 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 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 listAllSessions() { - return this.sessions.values(); - } - - @Override - public StoredMessage inFlightAck(String clientID, int messageID) { - LOG.debug("Acknowledging inflight message CId={}, messageId={}", clientID, messageID); - ConcurrentMap 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 inFlightForClient = this.inFlightIds.get(clientID); - if (inFlightForClient != null) { - inFlightForClient.remove(messageID); - } - return msg; - } - - @Override - public void inFlight(String clientID, int messageID, StoredMessage msg) { - ConcurrentMap 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 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 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 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 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 inflightPerClient = this.mvStore.openMap(inboundStoreForClient(clientID)); - if (inflightPerClient != null) { - totalInflight += inflightPerClient.size(); - } - -// totalInflight += countPubReleaseWaitingPubComplete(clientID); - - Map 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 sessionOlderThan(LocalDateTime queryPin) { - final Set 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; - } -}