From f44283d5f0e3397d05ea0753c6d617e2fe4c71c6 Mon Sep 17 00:00:00 2001 From: Dee Eldredge Date: Thu, 3 Mar 2016 16:08:52 -0800 Subject: [PATCH] Added onMessageConsumed logic. Related to: https://github.com/andsel/moquette/issues/156 --- .../AbstractInterceptHandler.java | 85 +-- .../interception/InterceptHandler.java | 101 ++-- .../io/moquette/interception/Interceptor.java | 96 ++-- .../messages/InterceptConsumedMessage.java | 15 + .../java/io/moquette/spi/ISessionsStore.java | 285 +++++----- .../moquette/spi/impl/BrokerInterceptor.java | 230 ++++---- .../spi/persistence/MapDBSessionsStore.java | 492 +++++++++--------- .../spi/impl/BrokerInterceptorTest.java | 245 ++++----- .../spi/persistence/MemorySessionStore.java | 476 ++++++++--------- 9 files changed, 1043 insertions(+), 982 deletions(-) create mode 100644 broker/src/main/java/io/moquette/interception/messages/InterceptConsumedMessage.java diff --git a/broker/src/main/java/io/moquette/interception/AbstractInterceptHandler.java b/broker/src/main/java/io/moquette/interception/AbstractInterceptHandler.java index 9b0fc56e..e39e0816 100644 --- a/broker/src/main/java/io/moquette/interception/AbstractInterceptHandler.java +++ b/broker/src/main/java/io/moquette/interception/AbstractInterceptHandler.java @@ -1,41 +1,44 @@ -/* - * Copyright (c) 2012-2015 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.interception; - -import io.moquette.interception.messages.*; - -/** - * Basic abstract class usefull to avoid empty methods creation in subclasses. - * - * Created by andrea on 08/12/15. - */ -public abstract class AbstractInterceptHandler implements InterceptHandler { - - @Override - public void onConnect(InterceptConnectMessage msg) {} - - @Override - public void onDisconnect(InterceptDisconnectMessage msg) {} - - @Override - public void onPublish(InterceptPublishMessage msg) {} - - @Override - public void onSubscribe(InterceptSubscribeMessage msg) {} - - @Override - public void onUnsubscribe(InterceptUnsubscribeMessage msg) {} -} +/* + * Copyright (c) 2012-2015 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.interception; + +import io.moquette.interception.messages.*; + +/** + * Basic abstract class usefull to avoid empty methods creation in subclasses. + * + * Created by andrea on 08/12/15. + */ +public abstract class AbstractInterceptHandler implements InterceptHandler { + + @Override + public void onConnect(InterceptConnectMessage msg) {} + + @Override + public void onDisconnect(InterceptDisconnectMessage msg) {} + + @Override + public void onPublish(InterceptPublishMessage msg) {} + + @Override + public void onSubscribe(InterceptSubscribeMessage msg) {} + + @Override + public void onUnsubscribe(InterceptUnsubscribeMessage msg) {} + + @Override + public void onMessageConsumed(InterceptConsumedMessage msg) {} +} diff --git a/broker/src/main/java/io/moquette/interception/InterceptHandler.java b/broker/src/main/java/io/moquette/interception/InterceptHandler.java index 397baf3f..434993e4 100644 --- a/broker/src/main/java/io/moquette/interception/InterceptHandler.java +++ b/broker/src/main/java/io/moquette/interception/InterceptHandler.java @@ -1,49 +1,52 @@ -/* - * Copyright (c) 2012-2015 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.interception; - -import io.moquette.parser.proto.messages.AbstractMessage; -import io.moquette.spi.impl.subscriptions.Subscription; -import io.moquette.interception.messages.InterceptConnectMessage; -import io.moquette.interception.messages.InterceptDisconnectMessage; -import io.moquette.interception.messages.InterceptPublishMessage; -import io.moquette.interception.messages.InterceptSubscribeMessage; -import io.moquette.interception.messages.InterceptUnsubscribeMessage; - -/** - * This interface is used to inject code for intercepting broker events. - *

- * The events can act only as observers. - *

- * Almost every method receives a subclass of {@link AbstractMessage}, except - * onDisconnect that receives the client id string and - * onSubscribe and onUnsubscribe that receive a - * {@link Subscription} object. - * - * @author Wagner Macedo - */ -public interface InterceptHandler { - - void onConnect(InterceptConnectMessage msg); - - void onDisconnect(InterceptDisconnectMessage msg); - - void onPublish(InterceptPublishMessage msg); - - void onSubscribe(InterceptSubscribeMessage msg); - - void onUnsubscribe(InterceptUnsubscribeMessage msg); -} +/* + * Copyright (c) 2012-2015 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.interception; + +import io.moquette.parser.proto.messages.AbstractMessage; +import io.moquette.spi.impl.subscriptions.Subscription; +import io.moquette.interception.messages.InterceptConnectMessage; +import io.moquette.interception.messages.InterceptConsumedMessage; +import io.moquette.interception.messages.InterceptDisconnectMessage; +import io.moquette.interception.messages.InterceptPublishMessage; +import io.moquette.interception.messages.InterceptSubscribeMessage; +import io.moquette.interception.messages.InterceptUnsubscribeMessage; + +/** + * This interface is used to inject code for intercepting broker events. + *

+ * The events can act only as observers. + *

+ * Almost every method receives a subclass of {@link AbstractMessage}, except + * onDisconnect that receives the client id string and + * onSubscribe and onUnsubscribe that receive a + * {@link Subscription} object. + * + * @author Wagner Macedo + */ +public interface InterceptHandler { + + void onConnect(InterceptConnectMessage msg); + + void onDisconnect(InterceptDisconnectMessage msg); + + void onPublish(InterceptPublishMessage msg); + + void onSubscribe(InterceptSubscribeMessage msg); + + void onUnsubscribe(InterceptUnsubscribeMessage msg); + + void onMessageConsumed(InterceptConsumedMessage msg); +} diff --git a/broker/src/main/java/io/moquette/interception/Interceptor.java b/broker/src/main/java/io/moquette/interception/Interceptor.java index 47214112..6687bea8 100644 --- a/broker/src/main/java/io/moquette/interception/Interceptor.java +++ b/broker/src/main/java/io/moquette/interception/Interceptor.java @@ -1,47 +1,49 @@ -/* - * Copyright (c) 2012-2015 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.interception; - -import io.moquette.parser.proto.messages.ConnectMessage; -import io.moquette.parser.proto.messages.PublishMessage; -import io.moquette.spi.impl.subscriptions.Subscription; - -/** - * This interface is to be used internally by the broker components. - *

- * An interface is used instead of a class to allow more flexibility in changing - * an implementation. - *

- * Interceptor implementations forward notifications to a InterceptHandler, - * that is normally a field. So, the implementations should act as a proxy to a custom - * intercept handler. - * - * @see InterceptHandler - * @author Wagner Macedo - */ -public interface Interceptor { - - void notifyClientConnected(ConnectMessage msg); - - void notifyClientDisconnected(String clientID); - - void notifyTopicPublished(PublishMessage msg, String clientID); - - void notifyTopicSubscribed(Subscription sub); - - void notifyTopicUnsubscribed(String topic, String clientID); - -} +/* + * Copyright (c) 2012-2015 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.interception; + +import io.moquette.interception.messages.InterceptConsumedMessage; +import io.moquette.parser.proto.messages.ConnectMessage; +import io.moquette.parser.proto.messages.PublishMessage; +import io.moquette.spi.impl.subscriptions.Subscription; + +/** + * This interface is to be used internally by the broker components. + *

+ * An interface is used instead of a class to allow more flexibility in changing + * an implementation. + *

+ * Interceptor implementations forward notifications to a InterceptHandler, + * that is normally a field. So, the implementations should act as a proxy to a custom + * intercept handler. + * + * @see InterceptHandler + * @author Wagner Macedo + */ +public interface Interceptor { + + void notifyClientConnected(ConnectMessage msg); + + void notifyClientDisconnected(String clientID); + + void notifyTopicPublished(PublishMessage msg, String clientID); + + void notifyTopicSubscribed(Subscription sub); + + void notifyTopicUnsubscribed(String topic, String clientID); + + void notifyMessageConsumed(InterceptConsumedMessage msg); +} diff --git a/broker/src/main/java/io/moquette/interception/messages/InterceptConsumedMessage.java b/broker/src/main/java/io/moquette/interception/messages/InterceptConsumedMessage.java new file mode 100644 index 00000000..991ecdeb --- /dev/null +++ b/broker/src/main/java/io/moquette/interception/messages/InterceptConsumedMessage.java @@ -0,0 +1,15 @@ +package io.moquette.interception.messages; + +import static io.moquette.spi.IMessagesStore.StoredMessage; + +public class InterceptConsumedMessage { + final private StoredMessage msg; + + public InterceptConsumedMessage( final StoredMessage msg) { + this.msg = msg; + } + + public StoredMessage getMsg() { + return msg; + } +} diff --git a/broker/src/main/java/io/moquette/spi/ISessionsStore.java b/broker/src/main/java/io/moquette/spi/ISessionsStore.java index bee5d133..3b9accfb 100644 --- a/broker/src/main/java/io/moquette/spi/ISessionsStore.java +++ b/broker/src/main/java/io/moquette/spi/ISessionsStore.java @@ -1,141 +1,144 @@ -/* - * Copyright (c) 2012-2015 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.util.Collection; -import java.util.List; - -import io.moquette.spi.impl.subscriptions.Subscription; - -/** - * Store used to handle the persistence of the subscriptions tree. - * - * @author andrea - */ -public interface ISessionsStore { - - void updateCleanStatus(String clientID, boolean cleanSession); - - class ClientTopicCouple { - public final String topicFilter; - public final String clientID; - - public ClientTopicCouple(String clientID, String topicFilter) { - this.clientID = clientID; - this.topicFilter = topicFilter; - } - - @Override - public boolean equals(Object o) { - if (this == o) return true; - if (o == null || getClass() != o.getClass()) return false; - - ClientTopicCouple that = (ClientTopicCouple) o; - - if (topicFilter != null ? !topicFilter.equals(that.topicFilter) : that.topicFilter != null) return false; - return !(clientID != null ? !clientID.equals(that.clientID) : that.clientID != null); - } - - @Override - public int hashCode() { - int result = topicFilter != null ? topicFilter.hashCode() : 0; - result = 31 * result + (clientID != null ? clientID.hashCode() : 0); - return result; - } - - @Override - public String toString() { - return "ClientTopicCouple{" + - "topicFilter='" + topicFilter + '\'' + - ", clientID='" + clientID + '\'' + - '}'; - } - - } - - void initStore(); - - /** - * Add a new subscription to the session - */ - void addNewSubscription(Subscription newSubscription); - - /** - * Removed a specific subscription - * */ - void removeSubscription(String topic, String clientID); - - /** - * Remove all the subscriptions of the session - */ - void wipeSubscriptions(String sessionID); - - /** - * Return all topic filters to recreate the subscription tree. - * */ - List listAllSubscriptions(); - - /** - * @return the subscription stored by clientID and topicFilter, if any else null; - * */ - Subscription getSubscription(ClientTopicCouple couple); - - /** - * @return true iff there are subscriptions persisted with clientID - */ - boolean contains(String clientID); - - ClientSession createNewSession(String clientID, boolean cleanSession); - - /** - * @param clientID the client owning the session. - * @return the session for the given clientID, null if not found. - * */ - ClientSession sessionForClient(String clientID); - - void inFlightAck(String clientID, int messageID); - - /** - * Save the binding messageID, clientID <-> guid - * */ - void inFlight(String clientID, int messageID, String guid); - - /** - * Return the next valid packetIdentifier for the given client session. - * */ - int nextPacketID(String clientID); - - /** - * Store the guid to be later published. - * */ - void bindToDeliver(String guid, String clientID); - - /** - * List the guids for retained messages for the session - * */ - Collection enqueued(String clientID); - - /** - * Remove form the queue of stored messages for session. - * */ - void removeEnqueued(String clientID, String guid); - - void secondPhaseAcknowledged(String clientID, int messageID); - - void secondPhaseAckWaiting(String clientID, int messageID); - - String mapToGuid(String clientID, int messageID); -} +/* + * Copyright (c) 2012-2015 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.util.Collection; +import java.util.List; + +import io.moquette.spi.IMessagesStore.StoredMessage; +import io.moquette.spi.impl.subscriptions.Subscription; + +/** + * Store used to handle the persistence of the subscriptions tree. + * + * @author andrea + */ +public interface ISessionsStore { + + void updateCleanStatus(String clientID, boolean cleanSession); + + class ClientTopicCouple { + public final String topicFilter; + public final String clientID; + + public ClientTopicCouple(String clientID, String topicFilter) { + this.clientID = clientID; + this.topicFilter = topicFilter; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + + ClientTopicCouple that = (ClientTopicCouple) o; + + if (topicFilter != null ? !topicFilter.equals(that.topicFilter) : that.topicFilter != null) return false; + return !(clientID != null ? !clientID.equals(that.clientID) : that.clientID != null); + } + + @Override + public int hashCode() { + int result = topicFilter != null ? topicFilter.hashCode() : 0; + result = 31 * result + (clientID != null ? clientID.hashCode() : 0); + return result; + } + + @Override + public String toString() { + return "ClientTopicCouple{" + + "topicFilter='" + topicFilter + '\'' + + ", clientID='" + clientID + '\'' + + '}'; + } + + } + + void initStore(); + + /** + * Add a new subscription to the session + */ + void addNewSubscription(Subscription newSubscription); + + /** + * Removed a specific subscription + * */ + void removeSubscription(String topic, String clientID); + + /** + * Remove all the subscriptions of the session + */ + void wipeSubscriptions(String sessionID); + + /** + * Return all topic filters to recreate the subscription tree. + * */ + List listAllSubscriptions(); + + /** + * @return the subscription stored by clientID and topicFilter, if any else null; + * */ + Subscription getSubscription(ClientTopicCouple couple); + + /** + * @return true iff there are subscriptions persisted with clientID + */ + boolean contains(String clientID); + + ClientSession createNewSession(String clientID, boolean cleanSession); + + /** + * @param clientID the client owning the session. + * @return the session for the given clientID, null if not found. + * */ + ClientSession sessionForClient(String clientID); + + void inFlightAck(String clientID, int messageID); + + /** + * Save the binding messageID, clientID <-> guid + * */ + void inFlight(String clientID, int messageID, String guid); + + /** + * Return the next valid packetIdentifier for the given client session. + * */ + int nextPacketID(String clientID); + + /** + * Store the guid to be later published. + * */ + void bindToDeliver(String guid, String clientID); + + /** + * List the guids for retained messages for the session + * */ + Collection enqueued(String clientID); + + /** + * Remove form the queue of stored messages for session. + * */ + void removeEnqueued(String clientID, String guid); + + void secondPhaseAcknowledged(String clientID, int messageID); + + void secondPhaseAckWaiting(String clientID, int messageID); + + String mapToGuid(String clientID, int messageID); + + public StoredMessage getInflightMessage(String clientID, int messageID); +} diff --git a/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java b/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java index 890cb831..f4ea53b4 100644 --- a/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java +++ b/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java @@ -1,109 +1,121 @@ -/* - * Copyright (c) 2012-2015 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.impl; - -import io.moquette.interception.InterceptHandler; -import io.moquette.interception.Interceptor; -import io.moquette.interception.messages.*; -import io.moquette.parser.proto.messages.ConnectMessage; -import io.moquette.spi.impl.subscriptions.Subscription; -import io.moquette.parser.proto.messages.PublishMessage; - -import java.util.List; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; - -/** - * An interceptor that execute the interception tasks asynchronously. - * - * @author Wagner Macedo - */ -final class BrokerInterceptor implements Interceptor { - private final List handlers; - private final ExecutorService executor; - - BrokerInterceptor(List handlers) { - this.handlers = handlers; - executor = Executors.newFixedThreadPool(1); - } - - /** - * Shutdown graciously the executor service - */ - void stop() { - executor.shutdown(); - } - - @Override - public void notifyClientConnected(final ConnectMessage msg) { - for (final InterceptHandler handler : this.handlers) { - executor.execute(new Runnable() { - @Override - public void run() { - handler.onConnect(new InterceptConnectMessage(msg)); - } - }); - } - } - - @Override - public void notifyClientDisconnected(final String clientID) { - for (final InterceptHandler handler : this.handlers) { - executor.execute(new Runnable() { - @Override - public void run() { - handler.onDisconnect(new InterceptDisconnectMessage(clientID)); - } - }); - } - } - - @Override - public void notifyTopicPublished(final PublishMessage msg, final String clientID) { - for (final InterceptHandler handler : this.handlers) { - executor.execute(new Runnable() { - @Override - public void run() { - handler.onPublish(new InterceptPublishMessage(msg, clientID)); - } - }); - } - } - - @Override - public void notifyTopicSubscribed(final Subscription sub) { - for (final InterceptHandler handler : this.handlers) { - executor.execute(new Runnable() { - @Override - public void run() { - handler.onSubscribe(new InterceptSubscribeMessage(sub)); - } - }); - } - } - - @Override - public void notifyTopicUnsubscribed(final String topic, final String clientID) { - for (final InterceptHandler handler : this.handlers) { - executor.execute(new Runnable() { - @Override - public void run() { - handler.onUnsubscribe(new InterceptUnsubscribeMessage(topic, clientID)); - } - }); - } - } -} +/* + * Copyright (c) 2012-2015 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.impl; + +import io.moquette.interception.InterceptHandler; +import io.moquette.interception.Interceptor; +import io.moquette.interception.messages.*; +import io.moquette.parser.proto.messages.ConnectMessage; +import io.moquette.spi.impl.subscriptions.Subscription; +import io.moquette.parser.proto.messages.PublishMessage; + +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; + +/** + * An interceptor that execute the interception tasks asynchronously. + * + * @author Wagner Macedo + */ +final class BrokerInterceptor implements Interceptor { + private final List handlers; + private final ExecutorService executor; + + BrokerInterceptor(List handlers) { + this.handlers = handlers; + executor = Executors.newFixedThreadPool(1); + } + + /** + * Shutdown graciously the executor service + */ + void stop() { + executor.shutdown(); + } + + @Override + public void notifyClientConnected(final ConnectMessage msg) { + for (final InterceptHandler handler : this.handlers) { + executor.execute(new Runnable() { + @Override + public void run() { + handler.onConnect(new InterceptConnectMessage(msg)); + } + }); + } + } + + @Override + public void notifyClientDisconnected(final String clientID) { + for (final InterceptHandler handler : this.handlers) { + executor.execute(new Runnable() { + @Override + public void run() { + handler.onDisconnect(new InterceptDisconnectMessage(clientID)); + } + }); + } + } + + @Override + public void notifyTopicPublished(final PublishMessage msg, final String clientID) { + for (final InterceptHandler handler : this.handlers) { + executor.execute(new Runnable() { + @Override + public void run() { + handler.onPublish(new InterceptPublishMessage(msg, clientID)); + } + }); + } + } + + @Override + public void notifyTopicSubscribed(final Subscription sub) { + for (final InterceptHandler handler : this.handlers) { + executor.execute(new Runnable() { + @Override + public void run() { + handler.onSubscribe(new InterceptSubscribeMessage(sub)); + } + }); + } + } + + @Override + public void notifyTopicUnsubscribed(final String topic, final String clientID) { + for (final InterceptHandler handler : this.handlers) { + executor.execute(new Runnable() { + @Override + public void run() { + handler.onUnsubscribe(new InterceptUnsubscribeMessage(topic, clientID)); + } + }); + } + } + + @Override + public void notifyMessageConsumed( final InterceptConsumedMessage msg ) { + for (final InterceptHandler handler : this.handlers) { + executor.execute(new Runnable() { + @Override + public void run() { + handler.onMessageConsumed(msg); + } + }); + } + } +} diff --git a/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java b/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java index 5e82ba95..bcfad2d9 100644 --- a/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java +++ b/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java @@ -1,240 +1,252 @@ -/* - * Copyright (c) 2012-2015 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.persistence; - -import io.moquette.spi.ClientSession; -import io.moquette.spi.IMessagesStore; -import io.moquette.spi.ISessionsStore; -import io.moquette.spi.impl.Utils; -import io.moquette.spi.impl.subscriptions.Subscription; -import io.moquette.spi.persistence.MapDBPersistentStore.PersistentSession; -import org.mapdb.DB; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.util.*; -import java.util.concurrent.ConcurrentMap; - -/** - * ISessionsStore implementation backed by MapDB. - * - * @author andrea - */ -class MapDBSessionsStore implements ISessionsStore { - - private static final Logger LOG = LoggerFactory.getLogger(MapDBSessionsStore.class); - - //maps clientID->[MessageId -> guid] - private ConcurrentMap> m_inflightStore; - //map clientID <-> set of currently in flight packet identifiers - private Map> m_inFlightIds; - private ConcurrentMap m_persistentSessions; - //maps clientID->[guid*], insertion order cares, it's queue - private ConcurrentMap> m_enqueuedStore; - //maps clientID->[messageID*] - private ConcurrentMap> m_secondPhaseStore; - - private final DB m_db; - private final IMessagesStore m_messagesStore; - - MapDBSessionsStore(DB db, IMessagesStore messagesStore) { - m_db = db; - m_messagesStore = messagesStore; - } - - @Override - public void initStore() { - m_inflightStore = m_db.getHashMap("inflight"); - m_inFlightIds = m_db.getHashMap("inflightPacketIDs"); - m_persistentSessions = m_db.getHashMap("sessions"); - m_enqueuedStore = m_db.getHashMap("sessionQueue"); - m_secondPhaseStore = m_db.getHashMap("secondPhase"); - } - - @Override - public void addNewSubscription(Subscription newSubscription) { - LOG.debug("addNewSubscription invoked with subscription {}", newSubscription); - final String clientID = newSubscription.getClientId(); - m_db.getHashMap("subscriptions_" + clientID).put(newSubscription.getTopicFilter(), newSubscription); - - if (LOG.isTraceEnabled()) { - LOG.trace("subscriptions_{}: {}", clientID, m_db.getHashMap("subscriptions_" + clientID)); - } - } - - @Override - public void removeSubscription(String topicFilter, String clientID) { - LOG.debug("removeSubscription topic filter: {} for clientID: {}", topicFilter, clientID); - if (!m_db.exists("subscriptions_" + clientID)) { - return; - } - m_db.getHashMap("subscriptions_" + clientID).remove(topicFilter); - } - - @Override - public void wipeSubscriptions(String clientID) { - LOG.debug("wipeSubscriptions"); - if (LOG.isTraceEnabled()) { - LOG.trace("Subscription pre wipe: subscriptions_{}: {}", clientID, m_db.getHashMap("subscriptions_" + clientID)); - } - m_db.delete("subscriptions_" + clientID); - if (LOG.isTraceEnabled()) { - LOG.trace("Subscription post wipe: subscriptions_{}: {}", clientID, m_db.getHashMap("subscriptions_" + clientID)); - } - } - - @Override - public List listAllSubscriptions() { - final List allSubscriptions = new ArrayList<>(); - for (String clientID : m_persistentSessions.keySet()) { - ConcurrentMap clientSubscriptions = m_db.getHashMap("subscriptions_" + clientID); - for (String topicFilter : clientSubscriptions.keySet()) { - allSubscriptions.add(new ClientTopicCouple(clientID, topicFilter)); - } - } - LOG.debug("retrieveAllSubscriptions returning subs {}", allSubscriptions); - return allSubscriptions; - } - - @Override - public Subscription getSubscription(ClientTopicCouple couple) { - ConcurrentMap clientSubscriptions = m_db.getHashMap("subscriptions_" + couple.clientID); - LOG.debug("subscriptions_{}: {}", couple.clientID, clientSubscriptions); - return clientSubscriptions.get(couple.topicFilter); - } - - @Override - public boolean contains(String clientID) { - return m_db.exists("subscriptions_" + clientID); - } - - @Override - public ClientSession createNewSession(String clientID, boolean cleanSession) { - LOG.debug("createNewSession for client <{}> with clean flag <{}>", clientID, cleanSession); - if (m_persistentSessions.containsKey(clientID)) { - LOG.error("already exists a session for client <{}>, bad condition", clientID); - throw new IllegalArgumentException("Can't create a session with the ID of an already existing" + clientID); - } - LOG.debug("clientID {} is a newcome, creating it's empty subscriptions set", clientID); - m_persistentSessions.putIfAbsent(clientID, new PersistentSession(cleanSession)); - return new ClientSession(clientID, m_messagesStore, this, cleanSession); - } - - @Override - public ClientSession sessionForClient(String clientID) { - if (!m_persistentSessions.containsKey(clientID)) { - return null; - } - - PersistentSession storedSession = m_persistentSessions.get(clientID); - return new ClientSession(clientID, m_messagesStore, this, storedSession.cleanSession); - } - - @Override - public void updateCleanStatus(String clientID, boolean cleanSession) { - m_persistentSessions.put(clientID, new MapDBPersistentStore.PersistentSession(cleanSession)); - } - - /** - * Return the next valid packetIdentifier for the given client session. - * */ - @Override - public int nextPacketID(String clientID) { - Set inFlightForClient = this.m_inFlightIds.get(clientID); - if (inFlightForClient == null) { - int nextPacketId = 1; - inFlightForClient = new HashSet<>(); - inFlightForClient.add(nextPacketId); - this.m_inFlightIds.put(clientID, inFlightForClient); - return nextPacketId; - - } - - int maxId = inFlightForClient.isEmpty() ? 0 : Collections.max(inFlightForClient); - int nextPacketId = (maxId + 1) % 0xFFFF; - inFlightForClient.add(nextPacketId); - return nextPacketId; - } - - @Override - public void inFlightAck(String clientID, int messageID) { - Map m = this.m_inflightStore.get(clientID); - if (m == null) { - LOG.error("Can't find the inFlight record for client <{}>", clientID); - return; - } - m.remove(messageID); - - //remove from the ids store - Set inFlightForClient = this.m_inFlightIds.get(clientID); - if (inFlightForClient != null) { - inFlightForClient.remove(messageID); - } - } - - @Override - public void inFlight(String clientID, int messageID, String guid) { - Map m = this.m_inflightStore.get(clientID); - if (m == null) { - m = new HashMap<>(); - } - m.put(messageID, guid); - this.m_inflightStore.put(clientID, m); - } - - @Override - public void bindToDeliver(String guid, String clientID) { - List guids = Utils.defaultGet(m_enqueuedStore, clientID, new ArrayList()); - guids.add(guid); - m_enqueuedStore.put(clientID, guids); - } - - @Override - public Collection enqueued(String clientID) { - return Utils.defaultGet(m_enqueuedStore, clientID, new ArrayList()); - } - - @Override - public void removeEnqueued(String clientID, String guid) { - List guids = Utils.defaultGet(m_enqueuedStore, clientID, new ArrayList()); - guids.remove(guid); - m_enqueuedStore.put(clientID, guids); - } - - @Override - public void secondPhaseAcknowledged(String clientID, int messageID) { - Set messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, new HashSet()); - messageIDs.remove(messageID); - m_secondPhaseStore.put(clientID, messageIDs); - } - - @Override - public void secondPhaseAckWaiting(String clientID, int messageID) { - Set messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, new HashSet()); - messageIDs.add(messageID); - m_secondPhaseStore.put(clientID, messageIDs); - } - - @Override - public String mapToGuid(String clientID, int messageID) { - ConcurrentMap messageIdToGuid = m_db.getHashMap(messageId2GuidsMapName(clientID)); - return messageIdToGuid.get(messageID); - } - - static String messageId2GuidsMapName(String clientID) { - return "guidsMapping_" + clientID; - } -} +/* + * Copyright (c) 2012-2015 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.persistence; + +import io.moquette.spi.ClientSession; +import io.moquette.spi.IMessagesStore; +import io.moquette.spi.IMessagesStore.StoredMessage; +import io.moquette.spi.ISessionsStore; +import io.moquette.spi.impl.Utils; +import io.moquette.spi.impl.subscriptions.Subscription; +import io.moquette.spi.persistence.MapDBPersistentStore.PersistentSession; +import org.mapdb.DB; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.*; +import java.util.concurrent.ConcurrentMap; + +/** + * ISessionsStore implementation backed by MapDB. + * + * @author andrea + */ +class MapDBSessionsStore implements ISessionsStore { + + private static final Logger LOG = LoggerFactory.getLogger(MapDBSessionsStore.class); + + //maps clientID->[MessageId -> guid] + private ConcurrentMap> m_inflightStore; + //map clientID <-> set of currently in flight packet identifiers + private Map> m_inFlightIds; + private ConcurrentMap m_persistentSessions; + //maps clientID->[guid*], insertion order cares, it's queue + private ConcurrentMap> m_enqueuedStore; + //maps clientID->[messageID*] + private ConcurrentMap> m_secondPhaseStore; + + private final DB m_db; + private final IMessagesStore m_messagesStore; + + MapDBSessionsStore(DB db, IMessagesStore messagesStore) { + m_db = db; + m_messagesStore = messagesStore; + } + + @Override + public void initStore() { + m_inflightStore = m_db.getHashMap("inflight"); + m_inFlightIds = m_db.getHashMap("inflightPacketIDs"); + m_persistentSessions = m_db.getHashMap("sessions"); + m_enqueuedStore = m_db.getHashMap("sessionQueue"); + m_secondPhaseStore = m_db.getHashMap("secondPhase"); + } + + @Override + public void addNewSubscription(Subscription newSubscription) { + LOG.debug("addNewSubscription invoked with subscription {}", newSubscription); + final String clientID = newSubscription.getClientId(); + m_db.getHashMap("subscriptions_" + clientID).put(newSubscription.getTopicFilter(), newSubscription); + + if (LOG.isTraceEnabled()) { + LOG.trace("subscriptions_{}: {}", clientID, m_db.getHashMap("subscriptions_" + clientID)); + } + } + + @Override + public void removeSubscription(String topicFilter, String clientID) { + LOG.debug("removeSubscription topic filter: {} for clientID: {}", topicFilter, clientID); + if (!m_db.exists("subscriptions_" + clientID)) { + return; + } + m_db.getHashMap("subscriptions_" + clientID).remove(topicFilter); + } + + @Override + public void wipeSubscriptions(String clientID) { + LOG.debug("wipeSubscriptions"); + if (LOG.isTraceEnabled()) { + LOG.trace("Subscription pre wipe: subscriptions_{}: {}", clientID, m_db.getHashMap("subscriptions_" + clientID)); + } + m_db.delete("subscriptions_" + clientID); + if (LOG.isTraceEnabled()) { + LOG.trace("Subscription post wipe: subscriptions_{}: {}", clientID, m_db.getHashMap("subscriptions_" + clientID)); + } + } + + @Override + public List listAllSubscriptions() { + final List allSubscriptions = new ArrayList<>(); + for (String clientID : m_persistentSessions.keySet()) { + ConcurrentMap clientSubscriptions = m_db.getHashMap("subscriptions_" + clientID); + for (String topicFilter : clientSubscriptions.keySet()) { + allSubscriptions.add(new ClientTopicCouple(clientID, topicFilter)); + } + } + LOG.debug("retrieveAllSubscriptions returning subs {}", allSubscriptions); + return allSubscriptions; + } + + @Override + public Subscription getSubscription(ClientTopicCouple couple) { + ConcurrentMap clientSubscriptions = m_db.getHashMap("subscriptions_" + couple.clientID); + LOG.debug("subscriptions_{}: {}", couple.clientID, clientSubscriptions); + return clientSubscriptions.get(couple.topicFilter); + } + + @Override + public boolean contains(String clientID) { + return m_db.exists("subscriptions_" + clientID); + } + + @Override + public ClientSession createNewSession(String clientID, boolean cleanSession) { + LOG.debug("createNewSession for client <{}> with clean flag <{}>", clientID, cleanSession); + if (m_persistentSessions.containsKey(clientID)) { + LOG.error("already exists a session for client <{}>, bad condition", clientID); + throw new IllegalArgumentException("Can't create a session with the ID of an already existing" + clientID); + } + LOG.debug("clientID {} is a newcome, creating it's empty subscriptions set", clientID); + m_persistentSessions.putIfAbsent(clientID, new PersistentSession(cleanSession)); + return new ClientSession(clientID, m_messagesStore, this, cleanSession); + } + + @Override + public ClientSession sessionForClient(String clientID) { + if (!m_persistentSessions.containsKey(clientID)) { + return null; + } + + PersistentSession storedSession = m_persistentSessions.get(clientID); + return new ClientSession(clientID, m_messagesStore, this, storedSession.cleanSession); + } + + @Override + public void updateCleanStatus(String clientID, boolean cleanSession) { + m_persistentSessions.put(clientID, new MapDBPersistentStore.PersistentSession(cleanSession)); + } + + /** + * Return the next valid packetIdentifier for the given client session. + * */ + @Override + public int nextPacketID(String clientID) { + Set inFlightForClient = this.m_inFlightIds.get(clientID); + if (inFlightForClient == null) { + int nextPacketId = 1; + inFlightForClient = new HashSet<>(); + inFlightForClient.add(nextPacketId); + this.m_inFlightIds.put(clientID, inFlightForClient); + return nextPacketId; + + } + + int maxId = inFlightForClient.isEmpty() ? 0 : Collections.max(inFlightForClient); + int nextPacketId = (maxId + 1) % 0xFFFF; + inFlightForClient.add(nextPacketId); + return nextPacketId; + } + + @Override + public void inFlightAck(String clientID, int messageID) { + Map m = this.m_inflightStore.get(clientID); + if (m == null) { + LOG.error("Can't find the inFlight record for client <{}>", clientID); + return; + } + m.remove(messageID); + + //remove from the ids store + Set inFlightForClient = this.m_inFlightIds.get(clientID); + if (inFlightForClient != null) { + inFlightForClient.remove(messageID); + } + } + + @Override + public void inFlight(String clientID, int messageID, String guid) { + Map m = this.m_inflightStore.get(clientID); + if (m == null) { + m = new HashMap<>(); + } + m.put(messageID, guid); + this.m_inflightStore.put(clientID, m); + } + + @Override + public void bindToDeliver(String guid, String clientID) { + List guids = Utils.defaultGet(m_enqueuedStore, clientID, new ArrayList()); + guids.add(guid); + m_enqueuedStore.put(clientID, guids); + } + + @Override + public Collection enqueued(String clientID) { + return Utils.defaultGet(m_enqueuedStore, clientID, new ArrayList()); + } + + @Override + public void removeEnqueued(String clientID, String guid) { + List guids = Utils.defaultGet(m_enqueuedStore, clientID, new ArrayList()); + guids.remove(guid); + m_enqueuedStore.put(clientID, guids); + } + + @Override + public void secondPhaseAcknowledged(String clientID, int messageID) { + Set messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, new HashSet()); + messageIDs.remove(messageID); + m_secondPhaseStore.put(clientID, messageIDs); + } + + @Override + public void secondPhaseAckWaiting(String clientID, int messageID) { + Set messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, new HashSet()); + messageIDs.add(messageID); + m_secondPhaseStore.put(clientID, messageIDs); + } + + @Override + public String mapToGuid(String clientID, int messageID) { + ConcurrentMap messageIdToGuid = m_db.getHashMap(messageId2GuidsMapName(clientID)); + return messageIdToGuid.get(messageID); + } + + static String messageId2GuidsMapName(String clientID) { + return "guidsMapping_" + clientID; + } + + @Override + public StoredMessage getInflightMessage( String clientID, int messageID ) { + Map clientEntries = m_inflightStore.get( clientID ); + if( clientEntries == null ) + return null; + String guid = clientEntries.get( messageID ); + if( guid == null ) + return null; + return m_messagesStore.getMessageByGuid( guid ); + } +} diff --git a/broker/src/test/java/io/moquette/spi/impl/BrokerInterceptorTest.java b/broker/src/test/java/io/moquette/spi/impl/BrokerInterceptorTest.java index 1ade4212..aded6043 100644 --- a/broker/src/test/java/io/moquette/spi/impl/BrokerInterceptorTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/BrokerInterceptorTest.java @@ -1,120 +1,125 @@ -/* - * Copyright (c) 2012-2015 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.impl; - -import io.moquette.interception.messages.*; -import io.moquette.parser.proto.messages.ConnectMessage; -import io.moquette.interception.InterceptHandler; -import io.moquette.parser.proto.messages.AbstractMessage; -import io.moquette.parser.proto.messages.PublishMessage; -import io.moquette.spi.impl.subscriptions.Subscription; -import org.junit.AfterClass; -import org.junit.BeforeClass; -import org.junit.Test; - -import java.util.Arrays; -import java.util.concurrent.atomic.AtomicInteger; - -import static org.junit.Assert.*; - -/** - * @author Wagner Macedo - */ -public class BrokerInterceptorTest { - // value to check for changes after every notification - private static final AtomicInteger n = new AtomicInteger(0); - - // Interceptor loaded with a custom InterceptHandler special for the tests - private static final class MockObserver implements InterceptHandler { - @Override - public void onConnect(InterceptConnectMessage msg) { - n.set(40); - } - - @Override - public void onDisconnect(InterceptDisconnectMessage msg) { - n.set(50); - } - - @Override - public void onPublish(InterceptPublishMessage msg) { - n.set(60); - } - - @Override - public void onSubscribe(InterceptSubscribeMessage msg) { - n.set(70); - } - - @Override - public void onUnsubscribe(InterceptUnsubscribeMessage msg) { - n.set(80); - } - } - - private static final BrokerInterceptor interceptor = new BrokerInterceptor(Arrays.asList(new MockObserver())); - - @BeforeClass - public static void beforeAllTests() { - // check if any of the handler methods was called before notifications - assertEquals(0, n.get()); - } - - @AfterClass - public static void afterAllTests() { - interceptor.stop(); - } - - /* Used to wait handler notification by the interceptor internal thread */ - private static void interval() throws InterruptedException { - Thread.sleep(100); - } - - @Test - public void testNotifyClientConnected() throws Exception { - interceptor.notifyClientConnected(new ConnectMessage()); - interval(); - assertEquals(40, n.get()); - } - - @Test - public void testNotifyClientDisconnected() throws Exception { - interceptor.notifyClientDisconnected("cli1234"); - interval(); - assertEquals(50, n.get()); - } - - @Test - public void testNotifyTopicPublished() throws Exception { - interceptor.notifyTopicPublished(new PublishMessage(), "cli1234"); - interval(); - assertEquals(60, n.get()); - } - - @Test - public void testNotifyTopicSubscribed() throws Exception { - interceptor.notifyTopicSubscribed(new Subscription("cli1", "o2", AbstractMessage.QOSType.MOST_ONE)); - interval(); - assertEquals(70, n.get()); - } - - @Test - public void testNotifyTopicUnsubscribed() throws Exception { - interceptor.notifyTopicUnsubscribed("o2", "cli1234"); - interval(); - assertEquals(80, n.get()); - } -} +/* + * Copyright (c) 2012-2015 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.impl; + +import io.moquette.interception.messages.*; +import io.moquette.parser.proto.messages.ConnectMessage; +import io.moquette.interception.InterceptHandler; +import io.moquette.parser.proto.messages.AbstractMessage; +import io.moquette.parser.proto.messages.PublishMessage; +import io.moquette.spi.impl.subscriptions.Subscription; +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.Test; + +import java.util.Arrays; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.Assert.*; + +/** + * @author Wagner Macedo + */ +public class BrokerInterceptorTest { + // value to check for changes after every notification + private static final AtomicInteger n = new AtomicInteger(0); + + // Interceptor loaded with a custom InterceptHandler special for the tests + private static final class MockObserver implements InterceptHandler { + @Override + public void onConnect(InterceptConnectMessage msg) { + n.set(40); + } + + @Override + public void onDisconnect(InterceptDisconnectMessage msg) { + n.set(50); + } + + @Override + public void onPublish(InterceptPublishMessage msg) { + n.set(60); + } + + @Override + public void onSubscribe(InterceptSubscribeMessage msg) { + n.set(70); + } + + @Override + public void onUnsubscribe(InterceptUnsubscribeMessage msg) { + n.set(80); + } + + @Override + public void onMessageConsumed( InterceptConsumedMessage msg ) { + n.set(90); + } + } + + private static final BrokerInterceptor interceptor = new BrokerInterceptor(Arrays.asList(new MockObserver())); + + @BeforeClass + public static void beforeAllTests() { + // check if any of the handler methods was called before notifications + assertEquals(0, n.get()); + } + + @AfterClass + public static void afterAllTests() { + interceptor.stop(); + } + + /* Used to wait handler notification by the interceptor internal thread */ + private static void interval() throws InterruptedException { + Thread.sleep(100); + } + + @Test + public void testNotifyClientConnected() throws Exception { + interceptor.notifyClientConnected(new ConnectMessage()); + interval(); + assertEquals(40, n.get()); + } + + @Test + public void testNotifyClientDisconnected() throws Exception { + interceptor.notifyClientDisconnected("cli1234"); + interval(); + assertEquals(50, n.get()); + } + + @Test + public void testNotifyTopicPublished() throws Exception { + interceptor.notifyTopicPublished(new PublishMessage(), "cli1234"); + interval(); + assertEquals(60, n.get()); + } + + @Test + public void testNotifyTopicSubscribed() throws Exception { + interceptor.notifyTopicSubscribed(new Subscription("cli1", "o2", AbstractMessage.QOSType.MOST_ONE)); + interval(); + assertEquals(70, n.get()); + } + + @Test + public void testNotifyTopicUnsubscribed() throws Exception { + interceptor.notifyTopicUnsubscribed("o2", "cli1234"); + interval(); + assertEquals(80, n.get()); + } +} diff --git a/broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java b/broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java index 1a228198..d22963cc 100644 --- a/broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java +++ b/broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java @@ -1,235 +1,241 @@ -/* - * Copyright (c) 2012-2015 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.persistence; - -import io.moquette.spi.ClientSession; -import io.moquette.spi.IMessagesStore; -import io.moquette.spi.ISessionsStore; -import io.moquette.spi.impl.Utils; -import io.moquette.spi.impl.subscriptions.Subscription; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.util.*; - -/** - * - * @author andrea - */ -public class MemorySessionStore implements ISessionsStore { - private static final Logger LOG = LoggerFactory.getLogger(MemorySessionStore.class); - - private Map> m_persistentSubscriptions = new HashMap<>(); - - private Map m_persistentSessions = new HashMap<>(); - - //maps clientID->[MessageId -> guid] - private Map> m_inflightStore = new HashMap<>(); -// private Map> m_inflightIDs = new HashMap<>(); - //maps clientID->[guid*] - private Map> m_enqueuedStore = new HashMap<>(); - //maps clientID->[messageID*] - private Map> m_secondPhaseStore = new HashMap<>(); - - private Map> m_messageToGuids; - private final IMessagesStore m_messagesStore; - - public MemorySessionStore(IMessagesStore messagesStore, Map> messageToGuids) { - m_messageToGuids = messageToGuids; - this.m_messagesStore = messagesStore; - } - - @Override - public void removeSubscription(String topic, String clientID) { - LOG.debug("removeSubscription topic filter: {} for clientID: {}", topic, clientID); - if (!m_persistentSubscriptions.containsKey(clientID)) { - return; - } - Set clientSubscriptions = m_persistentSubscriptions.get(clientID); - //search for the subscription to remove - Subscription toBeRemoved = null; - for (Subscription sub : clientSubscriptions) { - if (sub.getTopicFilter().equals(topic)) { - toBeRemoved = sub; - break; - } - } - - if (toBeRemoved != null) { - clientSubscriptions.remove(toBeRemoved); - } - } - - @Override - public void initStore() { - - } - - @Override - public void addNewSubscription(Subscription newSubscription) { - final String clientID = newSubscription.getClientId(); - if (!m_persistentSubscriptions.containsKey(clientID)) { - m_persistentSubscriptions.put(clientID, new HashSet()); - } - - Set subs = m_persistentSubscriptions.get(clientID); - subs.remove(newSubscription); //same topic and clientID - subs.add(newSubscription); - m_persistentSubscriptions.put(clientID, subs); - } - - @Override - public void wipeSubscriptions(String clientID) { - m_persistentSubscriptions.remove(clientID); - } - - @Override - public boolean contains(String clientID) { - return m_persistentSubscriptions.containsKey(clientID); - } - - @Override - public ClientSession createNewSession(String clientID, boolean cleanSession) { - LOG.debug("createNewSession for client <{}>", clientID); - if (m_persistentSessions.containsKey(clientID)) { - LOG.error("already exists a session for client <{}>, bad condition", clientID); - throw new IllegalArgumentException("Can't create a session with the ID of an already existing" + clientID); - } - LOG.debug("clientID {} is a newcome, creating it's empty subscriptions set", clientID); - m_persistentSubscriptions.put(clientID, new HashSet()); - m_persistentSessions.put(clientID, new MapDBPersistentStore.PersistentSession(cleanSession)); - return new ClientSession(clientID, m_messagesStore, this, cleanSession); - } - - @Override - public ClientSession sessionForClient(String clientID) { - if (!m_persistentSessions.containsKey(clientID)) { - return null; - } - - MapDBPersistentStore.PersistentSession storedSession = m_persistentSessions.get(clientID); - return new ClientSession(clientID, m_messagesStore, this, storedSession.cleanSession); - } - - @Override - public void updateCleanStatus(String clientID, boolean cleanSession) { - m_persistentSessions.put(clientID, new MapDBPersistentStore.PersistentSession(cleanSession)); - } - - @Override - public List listAllSubscriptions() { - List allSubscriptions = new ArrayList<>(); - for (Map.Entry> entry : m_persistentSubscriptions.entrySet()) { - for (Subscription sub : entry.getValue()) { - allSubscriptions.add(sub.asClientTopicCouple()); - } - } - return allSubscriptions; - } - - @Override - public Subscription getSubscription(ClientTopicCouple couple) { - Set subscriptions = m_persistentSubscriptions.get(couple.clientID); - if (subscriptions == null || subscriptions.isEmpty()) { - return null; - } - for (Subscription sub : subscriptions) { - if (sub.getTopicFilter().equals(couple.topicFilter)) { - return sub; - } - } - return null; - } - - @Override - public void inFlightAck(String clientID, int messageID) { - Map m = this.m_inflightStore.get(clientID); - if (m == null) { - LOG.error("Can't find the inFlight record for client <{}>", clientID); - return; - } - m.remove(messageID); - } - - @Override - public void inFlight(String clientID, int messageID, String guid) { - Map m = this.m_inflightStore.get(clientID); - if (m == null) { - m = new HashMap<>(); - } - m.put(messageID, guid); - this.m_inflightStore.put(clientID, m); - } - - /** - * Return the next valid packetIdentifier for the given client session. - * */ - @Override - public int nextPacketID(String clientID) { - Map m = this.m_inflightStore.get(clientID); - if (m == null) { - m = new HashMap<>(); - int nextPacketId = 1; - m.put(nextPacketId, null); - return nextPacketId; - } - int maxId = Collections.max(m.keySet()); - int nextPacketId = (maxId + 1) % 0xFFFF; - m.put(nextPacketId, null); - return nextPacketId; - } - - @Override - public void bindToDeliver(String guid, String clientID) { - Set guids = Utils.defaultGet(m_enqueuedStore, clientID, new HashSet()); - guids.add(guid); - m_enqueuedStore.put(clientID, guids); - } - - @Override - public Collection enqueued(String clientID) { - return Utils.defaultGet(m_enqueuedStore, clientID, new HashSet()); - } - - @Override - public void removeEnqueued(String clientID, String guid) { - Set guids = Utils.defaultGet(m_enqueuedStore, clientID, new HashSet()); - guids.remove(guid); - m_enqueuedStore.put(clientID, guids); - } - - @Override - public void secondPhaseAcknowledged(String clientID, int messageID) { - Set messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, new HashSet()); - messageIDs.remove(messageID); - m_secondPhaseStore.put(clientID, messageIDs); - } - - @Override - public void secondPhaseAckWaiting(String clientID, int messageID) { - Set messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, new HashSet()); - messageIDs.add(messageID); - m_secondPhaseStore.put(clientID, messageIDs); - } - - @Override - public String mapToGuid(String clientID, int messageID) { - HashMap guids = (HashMap) Utils.defaultGet(m_messageToGuids, - clientID, new HashMap()); - return guids.get(messageID); - } -} +/* + * Copyright (c) 2012-2015 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.persistence; + +import io.moquette.spi.ClientSession; +import io.moquette.spi.IMessagesStore; +import io.moquette.spi.IMessagesStore.StoredMessage; +import io.moquette.spi.ISessionsStore; +import io.moquette.spi.impl.Utils; +import io.moquette.spi.impl.subscriptions.Subscription; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.*; + +/** + * + * @author andrea + */ +public class MemorySessionStore implements ISessionsStore { + private static final Logger LOG = LoggerFactory.getLogger(MemorySessionStore.class); + + private Map> m_persistentSubscriptions = new HashMap<>(); + + private Map m_persistentSessions = new HashMap<>(); + + //maps clientID->[MessageId -> guid] + private Map> m_inflightStore = new HashMap<>(); +// private Map> m_inflightIDs = new HashMap<>(); + //maps clientID->[guid*] + private Map> m_enqueuedStore = new HashMap<>(); + //maps clientID->[messageID*] + private Map> m_secondPhaseStore = new HashMap<>(); + + private Map> m_messageToGuids; + private final IMessagesStore m_messagesStore; + + public MemorySessionStore(IMessagesStore messagesStore, Map> messageToGuids) { + m_messageToGuids = messageToGuids; + this.m_messagesStore = messagesStore; + } + + @Override + public void removeSubscription(String topic, String clientID) { + LOG.debug("removeSubscription topic filter: {} for clientID: {}", topic, clientID); + if (!m_persistentSubscriptions.containsKey(clientID)) { + return; + } + Set clientSubscriptions = m_persistentSubscriptions.get(clientID); + //search for the subscription to remove + Subscription toBeRemoved = null; + for (Subscription sub : clientSubscriptions) { + if (sub.getTopicFilter().equals(topic)) { + toBeRemoved = sub; + break; + } + } + + if (toBeRemoved != null) { + clientSubscriptions.remove(toBeRemoved); + } + } + + @Override + public void initStore() { + + } + + @Override + public void addNewSubscription(Subscription newSubscription) { + final String clientID = newSubscription.getClientId(); + if (!m_persistentSubscriptions.containsKey(clientID)) { + m_persistentSubscriptions.put(clientID, new HashSet()); + } + + Set subs = m_persistentSubscriptions.get(clientID); + subs.remove(newSubscription); //same topic and clientID + subs.add(newSubscription); + m_persistentSubscriptions.put(clientID, subs); + } + + @Override + public void wipeSubscriptions(String clientID) { + m_persistentSubscriptions.remove(clientID); + } + + @Override + public boolean contains(String clientID) { + return m_persistentSubscriptions.containsKey(clientID); + } + + @Override + public ClientSession createNewSession(String clientID, boolean cleanSession) { + LOG.debug("createNewSession for client <{}>", clientID); + if (m_persistentSessions.containsKey(clientID)) { + LOG.error("already exists a session for client <{}>, bad condition", clientID); + throw new IllegalArgumentException("Can't create a session with the ID of an already existing" + clientID); + } + LOG.debug("clientID {} is a newcome, creating it's empty subscriptions set", clientID); + m_persistentSubscriptions.put(clientID, new HashSet()); + m_persistentSessions.put(clientID, new MapDBPersistentStore.PersistentSession(cleanSession)); + return new ClientSession(clientID, m_messagesStore, this, cleanSession); + } + + @Override + public ClientSession sessionForClient(String clientID) { + if (!m_persistentSessions.containsKey(clientID)) { + return null; + } + + MapDBPersistentStore.PersistentSession storedSession = m_persistentSessions.get(clientID); + return new ClientSession(clientID, m_messagesStore, this, storedSession.cleanSession); + } + + @Override + public void updateCleanStatus(String clientID, boolean cleanSession) { + m_persistentSessions.put(clientID, new MapDBPersistentStore.PersistentSession(cleanSession)); + } + + @Override + public List listAllSubscriptions() { + List allSubscriptions = new ArrayList<>(); + for (Map.Entry> entry : m_persistentSubscriptions.entrySet()) { + for (Subscription sub : entry.getValue()) { + allSubscriptions.add(sub.asClientTopicCouple()); + } + } + return allSubscriptions; + } + + @Override + public Subscription getSubscription(ClientTopicCouple couple) { + Set subscriptions = m_persistentSubscriptions.get(couple.clientID); + if (subscriptions == null || subscriptions.isEmpty()) { + return null; + } + for (Subscription sub : subscriptions) { + if (sub.getTopicFilter().equals(couple.topicFilter)) { + return sub; + } + } + return null; + } + + @Override + public void inFlightAck(String clientID, int messageID) { + Map m = this.m_inflightStore.get(clientID); + if (m == null) { + LOG.error("Can't find the inFlight record for client <{}>", clientID); + return; + } + m.remove(messageID); + } + + @Override + public void inFlight(String clientID, int messageID, String guid) { + Map m = this.m_inflightStore.get(clientID); + if (m == null) { + m = new HashMap<>(); + } + m.put(messageID, guid); + this.m_inflightStore.put(clientID, m); + } + + /** + * Return the next valid packetIdentifier for the given client session. + * */ + @Override + public int nextPacketID(String clientID) { + Map m = this.m_inflightStore.get(clientID); + if (m == null) { + m = new HashMap<>(); + int nextPacketId = 1; + m.put(nextPacketId, null); + return nextPacketId; + } + int maxId = Collections.max(m.keySet()); + int nextPacketId = (maxId + 1) % 0xFFFF; + m.put(nextPacketId, null); + return nextPacketId; + } + + @Override + public void bindToDeliver(String guid, String clientID) { + Set guids = Utils.defaultGet(m_enqueuedStore, clientID, new HashSet()); + guids.add(guid); + m_enqueuedStore.put(clientID, guids); + } + + @Override + public Collection enqueued(String clientID) { + return Utils.defaultGet(m_enqueuedStore, clientID, new HashSet()); + } + + @Override + public void removeEnqueued(String clientID, String guid) { + Set guids = Utils.defaultGet(m_enqueuedStore, clientID, new HashSet()); + guids.remove(guid); + m_enqueuedStore.put(clientID, guids); + } + + @Override + public void secondPhaseAcknowledged(String clientID, int messageID) { + Set messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, new HashSet()); + messageIDs.remove(messageID); + m_secondPhaseStore.put(clientID, messageIDs); + } + + @Override + public void secondPhaseAckWaiting(String clientID, int messageID) { + Set messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, new HashSet()); + messageIDs.add(messageID); + m_secondPhaseStore.put(clientID, messageIDs); + } + + @Override + public String mapToGuid(String clientID, int messageID) { + HashMap guids = (HashMap) Utils.defaultGet(m_messageToGuids, + clientID, new HashMap()); + return guids.get(messageID); + } + + @Override + public StoredMessage getInflightMessage( String clientID, int messageID ) { + return null; + } +}