From ec6ec09ceec846c9334d4d32a1e19b9857c5ec16 Mon Sep 17 00:00:00 2001 From: andsel Date: Sun, 19 Mar 2017 22:01:23 +0100 Subject: [PATCH] Minor fixed comment (#281) --- .../java/io/moquette/BrokerConstants.java | 2 +- .../AbstractInterceptHandler.java | 132 ++-- .../interception/InterceptHandler.java | 136 ++-- .../io/moquette/interception/Interceptor.java | 108 +-- .../messages/InterceptAbstractMessage.java | 2 +- .../moquette/server/ConnectionDescriptor.java | 2 +- .../java/io/moquette/server/Constants.java | 2 +- .../DefaultMoquetteSslContextCreator.java | 2 +- .../main/java/io/moquette/server/Server.java | 2 +- .../io/moquette/server/ServerAcceptor.java | 2 +- .../server/config/ConfigurationParser.java | 2 +- .../io/moquette/server/config/IConfig.java | 2 +- .../moquette/server/config/MemoryConfig.java | 2 +- .../server/config/ResourceLoaderConfig.java | 2 +- .../moquette/server/netty/MessageBuilder.java | 2 +- .../netty/MoquetteIdleTimeoutHandler.java | 2 +- .../moquette/server/netty/NettyAcceptor.java | 2 +- .../server/netty/NettyMQTTHandler.java | 2 +- .../io/moquette/server/netty/NettyUtils.java | 2 +- .../server/netty/metrics/BytesMetrics.java | 2 +- .../netty/metrics/BytesMetricsCollector.java | 2 +- .../netty/metrics/BytesMetricsHandler.java | 2 +- .../netty/metrics/MQTTMessageLogger.java | 2 +- .../server/netty/metrics/MessageMetrics.java | 2 +- .../metrics/MessageMetricsCollector.java | 2 +- .../netty/metrics/MessageMetricsHandler.java | 2 +- .../java/io/moquette/spi/ClientSession.java | 2 +- .../io/moquette/spi/IMatchingCondition.java | 2 +- .../java/io/moquette/spi/IMessagesStore.java | 2 +- .../main/java/io/moquette/spi/IMessaging.java | 2 +- .../java/io/moquette/spi/ISessionsStore.java | 426 +++++------ .../java/io/moquette/spi/MessageGUID.java | 2 +- .../moquette/spi/impl/BrokerInterceptor.java | 520 ++++++------- .../java/io/moquette/spi/impl/DebugUtils.java | 2 +- .../moquette/spi/impl/ProtocolProcessor.java | 2 +- .../impl/ProtocolProcessorBootstrapper.java | 2 +- .../spi/impl/security/ACLFileParser.java | 2 +- .../spi/impl/security/Authorization.java | 2 +- .../security/AuthorizationsCollector.java | 2 +- .../impl/security/DenyAllAuthorizator.java | 2 +- .../spi/impl/security/FileAuthenticator.java | 2 +- .../impl/security/PermitAllAuthorizator.java | 2 +- .../impl/security/ResourceAuthenticator.java | 2 +- .../spi/impl/subscriptions/Subscription.java | 2 +- .../subscriptions/SubscriptionsStore.java | 2 +- .../spi/impl/subscriptions/Token.java | 2 +- .../spi/impl/subscriptions/Topic.java | 2 +- .../spi/impl/subscriptions/TreeNode.java | 2 +- .../spi/persistence/MapDBMessagesStore.java | 2 +- .../spi/persistence/MapDBPersistentStore.java | 2 +- .../spi/persistence/MapDBSessionsStore.java | 696 +++++++++--------- .../moquette/spi/security/IAuthenticator.java | 2 +- .../moquette/spi/security/IAuthorizator.java | 2 +- .../spi/security/ISslContextCreator.java | 2 +- broker/src/test/java/io/moquette/Utils.java | 2 +- .../server/ConfigurationClassLoaderTest.java | 2 +- .../io/moquette/server/IntegrationUtils.java | 2 +- .../io/moquette/server/MQTTWebSocket.java | 2 +- .../ServerIntegrationEmbeddedPublishTest.java | 2 +- .../server/ServerIntegrationFuseTest.java | 2 +- .../server/ServerIntegrationPahoTest.java | 2 +- .../ServerIntegrationQoSValidationTest.java | 2 +- .../ServerIntegrationSSLClientAuthTest.java | 2 +- .../server/ServerIntegrationSSLTest.java | 2 +- .../ServerIntegrationWebSocketTest.java | 2 +- ...erverLowlevelMessagesIntegrationTests.java | 2 +- .../server/ServerRestartIntegrationTest.java | 2 +- .../config/ConfigurationParserTest.java | 2 +- .../spec/v3_1_1/connection/ConnectionIT.java | 2 +- .../spi/impl/BrokerInterceptorTest.java | 348 ++++----- .../spi/impl/MemoryMessagesStore.java | 2 +- .../spi/impl/MemoryStorageService.java | 2 +- .../moquette/spi/impl/MockAuthenticator.java | 2 +- .../spi/impl/MockReceiverChannel.java | 2 +- .../spi/impl/NettyChannelAssertions.java | 2 +- .../spi/impl/ProtocolProcessorTest.java | 2 +- .../impl/ProtocolProcessor_CONNECT_Test.java | 2 +- .../spi/impl/security/ACLFileParserTest.java | 2 +- .../security/AuthorizationsCollectorTest.java | 2 +- .../impl/security/FileAuthenticatorTest.java | 2 +- .../subscriptions/SubscriptionsStoreTest.java | 2 +- .../spi/impl/subscriptions/TopicTest.java | 2 +- .../persistence/MapDBPersistentStoreTest.java | 2 +- .../spi/persistence/MemorySessionStore.java | 546 +++++++------- .../java/io/moquette/testclient/Client.java | 2 +- .../testclient/ClientNettyMQTTHandler.java | 2 +- .../testembedded/EmbeddedLauncher.java | 4 +- 87 files changed, 1536 insertions(+), 1536 deletions(-) diff --git a/broker/src/main/java/io/moquette/BrokerConstants.java b/broker/src/main/java/io/moquette/BrokerConstants.java index 61d605ba..12091938 100644 --- a/broker/src/main/java/io/moquette/BrokerConstants.java +++ b/broker/src/main/java/io/moquette/BrokerConstants.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/main/java/io/moquette/interception/AbstractInterceptHandler.java b/broker/src/main/java/io/moquette/interception/AbstractInterceptHandler.java index 6c99ef51..2d80a4fa 100644 --- a/broker/src/main/java/io/moquette/interception/AbstractInterceptHandler.java +++ b/broker/src/main/java/io/moquette/interception/AbstractInterceptHandler.java @@ -1,66 +1,66 @@ -/* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() - * ------------------------------------------------------ - * 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.InterceptAcknowledgedMessage; -import io.moquette.interception.messages.InterceptConnectMessage; -import io.moquette.interception.messages.InterceptConnectionLostMessage; -import io.moquette.interception.messages.InterceptDisconnectMessage; -import io.moquette.interception.messages.InterceptPublishMessage; -import io.moquette.interception.messages.InterceptSubscribeMessage; -import io.moquette.interception.messages.InterceptUnsubscribeMessage; - -/** - * 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 Class[] getInterceptedMessageTypes() { - return InterceptHandler.ALL_MESSAGE_TYPES; - } - - @Override - public void onConnect(InterceptConnectMessage msg) { - } - - @Override - public void onDisconnect(InterceptDisconnectMessage msg) { - } - - @Override - public void onConnectionLost(InterceptConnectionLostMessage msg) { - } - - @Override - public void onPublish(InterceptPublishMessage msg) { - } - - @Override - public void onSubscribe(InterceptSubscribeMessage msg) { - } - - @Override - public void onUnsubscribe(InterceptUnsubscribeMessage msg) { - } - - @Override - public void onMessageAcknowledged(InterceptAcknowledgedMessage msg) { - } -} +/* + * Copyright (c) 2012-2017 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.InterceptAcknowledgedMessage; +import io.moquette.interception.messages.InterceptConnectMessage; +import io.moquette.interception.messages.InterceptConnectionLostMessage; +import io.moquette.interception.messages.InterceptDisconnectMessage; +import io.moquette.interception.messages.InterceptPublishMessage; +import io.moquette.interception.messages.InterceptSubscribeMessage; +import io.moquette.interception.messages.InterceptUnsubscribeMessage; + +/** + * 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 Class[] getInterceptedMessageTypes() { + return InterceptHandler.ALL_MESSAGE_TYPES; + } + + @Override + public void onConnect(InterceptConnectMessage msg) { + } + + @Override + public void onDisconnect(InterceptDisconnectMessage msg) { + } + + @Override + public void onConnectionLost(InterceptConnectionLostMessage msg) { + } + + @Override + public void onPublish(InterceptPublishMessage msg) { + } + + @Override + public void onSubscribe(InterceptSubscribeMessage msg) { + } + + @Override + public void onUnsubscribe(InterceptUnsubscribeMessage msg) { + } + + @Override + public void onMessageAcknowledged(InterceptAcknowledgedMessage msg) { + } +} diff --git a/broker/src/main/java/io/moquette/interception/InterceptHandler.java b/broker/src/main/java/io/moquette/interception/InterceptHandler.java index 064b74e4..db43d7ea 100644 --- a/broker/src/main/java/io/moquette/interception/InterceptHandler.java +++ b/broker/src/main/java/io/moquette/interception/InterceptHandler.java @@ -1,68 +1,68 @@ -/* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() - * ------------------------------------------------------ - * 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.*; -import io.moquette.spi.impl.subscriptions.Subscription; -import io.netty.handler.codec.mqtt.MqttMessage; - -/** - * 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 MqttMessage}, except onDisconnect - * that receives the client id string and onSubscribe and onUnsubscribe - * that receive a {@link Subscription} object. - * - * @author Wagner Macedo - */ -public interface InterceptHandler { - - Class[] ALL_MESSAGE_TYPES = {InterceptConnectMessage.class, InterceptDisconnectMessage.class, - InterceptConnectionLostMessage.class, InterceptPublishMessage.class, InterceptSubscribeMessage.class, - InterceptUnsubscribeMessage.class, InterceptAcknowledgedMessage.class}; - - /** - * Returns the identifier of this intercept handler. - * - * @return - */ - String getID(); - - /** - * Returns the InterceptMessage subtypes that this handler can process. If the result is null or - * equal to ALL_MESSAGE_TYPES, all the message types will be processed. - * - * @return - */ - Class[] getInterceptedMessageTypes(); - - void onConnect(InterceptConnectMessage msg); - - void onDisconnect(InterceptDisconnectMessage msg); - - void onConnectionLost(InterceptConnectionLostMessage msg); - - void onPublish(InterceptPublishMessage msg); - - void onSubscribe(InterceptSubscribeMessage msg); - - void onUnsubscribe(InterceptUnsubscribeMessage msg); - - void onMessageAcknowledged(InterceptAcknowledgedMessage msg); -} +/* + * Copyright (c) 2012-2017 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.*; +import io.moquette.spi.impl.subscriptions.Subscription; +import io.netty.handler.codec.mqtt.MqttMessage; + +/** + * 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 MqttMessage}, except onDisconnect + * that receives the client id string and onSubscribe and onUnsubscribe + * that receive a {@link Subscription} object. + * + * @author Wagner Macedo + */ +public interface InterceptHandler { + + Class[] ALL_MESSAGE_TYPES = {InterceptConnectMessage.class, InterceptDisconnectMessage.class, + InterceptConnectionLostMessage.class, InterceptPublishMessage.class, InterceptSubscribeMessage.class, + InterceptUnsubscribeMessage.class, InterceptAcknowledgedMessage.class}; + + /** + * Returns the identifier of this intercept handler. + * + * @return + */ + String getID(); + + /** + * Returns the InterceptMessage subtypes that this handler can process. If the result is null or + * equal to ALL_MESSAGE_TYPES, all the message types will be processed. + * + * @return + */ + Class[] getInterceptedMessageTypes(); + + void onConnect(InterceptConnectMessage msg); + + void onDisconnect(InterceptDisconnectMessage msg); + + void onConnectionLost(InterceptConnectionLostMessage msg); + + void onPublish(InterceptPublishMessage msg); + + void onSubscribe(InterceptSubscribeMessage msg); + + void onUnsubscribe(InterceptUnsubscribeMessage msg); + + void onMessageAcknowledged(InterceptAcknowledgedMessage msg); +} diff --git a/broker/src/main/java/io/moquette/interception/Interceptor.java b/broker/src/main/java/io/moquette/interception/Interceptor.java index b41dc4b8..df1f8873 100644 --- a/broker/src/main/java/io/moquette/interception/Interceptor.java +++ b/broker/src/main/java/io/moquette/interception/Interceptor.java @@ -1,54 +1,54 @@ -/* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() - * ------------------------------------------------------ - * 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.InterceptAcknowledgedMessage; -import io.moquette.spi.impl.subscriptions.Subscription; -import io.netty.handler.codec.mqtt.MqttConnectMessage; -import io.netty.handler.codec.mqtt.MqttPublishMessage; - -/** - * 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(MqttConnectMessage msg); - - void notifyClientDisconnected(String clientID, String username); - - void notifyClientConnectionLost(String clientID, String username); - - void notifyTopicPublished(MqttPublishMessage msg, String clientID, final String username); - - void notifyTopicSubscribed(Subscription sub, final String username); - - void notifyTopicUnsubscribed(String topic, String clientID, final String username); - - void notifyMessageAcknowledged(InterceptAcknowledgedMessage msg); - - void addInterceptHandler(InterceptHandler interceptHandler); - - void removeInterceptHandler(InterceptHandler interceptHandler); -} +/* + * Copyright (c) 2012-2017 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.InterceptAcknowledgedMessage; +import io.moquette.spi.impl.subscriptions.Subscription; +import io.netty.handler.codec.mqtt.MqttConnectMessage; +import io.netty.handler.codec.mqtt.MqttPublishMessage; + +/** + * 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(MqttConnectMessage msg); + + void notifyClientDisconnected(String clientID, String username); + + void notifyClientConnectionLost(String clientID, String username); + + void notifyTopicPublished(MqttPublishMessage msg, String clientID, final String username); + + void notifyTopicSubscribed(Subscription sub, final String username); + + void notifyTopicUnsubscribed(String topic, String clientID, final String username); + + void notifyMessageAcknowledged(InterceptAcknowledgedMessage msg); + + void addInterceptHandler(InterceptHandler interceptHandler); + + void removeInterceptHandler(InterceptHandler interceptHandler); +} diff --git a/broker/src/main/java/io/moquette/interception/messages/InterceptAbstractMessage.java b/broker/src/main/java/io/moquette/interception/messages/InterceptAbstractMessage.java index 26d62fbb..4878e655 100644 --- a/broker/src/main/java/io/moquette/interception/messages/InterceptAbstractMessage.java +++ b/broker/src/main/java/io/moquette/interception/messages/InterceptAbstractMessage.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java b/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java index 9ea41bcf..47e9b2a3 100644 --- a/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java +++ b/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/main/java/io/moquette/server/Constants.java b/broker/src/main/java/io/moquette/server/Constants.java index 3a77379b..44a64d9f 100644 --- a/broker/src/main/java/io/moquette/server/Constants.java +++ b/broker/src/main/java/io/moquette/server/Constants.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/main/java/io/moquette/server/DefaultMoquetteSslContextCreator.java b/broker/src/main/java/io/moquette/server/DefaultMoquetteSslContextCreator.java index 16bec248..d8c2d27c 100644 --- a/broker/src/main/java/io/moquette/server/DefaultMoquetteSslContextCreator.java +++ b/broker/src/main/java/io/moquette/server/DefaultMoquetteSslContextCreator.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/main/java/io/moquette/server/Server.java b/broker/src/main/java/io/moquette/server/Server.java index 87074b32..b4b002c1 100644 --- a/broker/src/main/java/io/moquette/server/Server.java +++ b/broker/src/main/java/io/moquette/server/Server.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/main/java/io/moquette/server/ServerAcceptor.java b/broker/src/main/java/io/moquette/server/ServerAcceptor.java index b750715a..ec3a63a5 100644 --- a/broker/src/main/java/io/moquette/server/ServerAcceptor.java +++ b/broker/src/main/java/io/moquette/server/ServerAcceptor.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/main/java/io/moquette/server/config/ConfigurationParser.java b/broker/src/main/java/io/moquette/server/config/ConfigurationParser.java index 3b95573f..d7429e18 100644 --- a/broker/src/main/java/io/moquette/server/config/ConfigurationParser.java +++ b/broker/src/main/java/io/moquette/server/config/ConfigurationParser.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/main/java/io/moquette/server/config/IConfig.java b/broker/src/main/java/io/moquette/server/config/IConfig.java index 55ebaeca..50096a94 100644 --- a/broker/src/main/java/io/moquette/server/config/IConfig.java +++ b/broker/src/main/java/io/moquette/server/config/IConfig.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/main/java/io/moquette/server/config/MemoryConfig.java b/broker/src/main/java/io/moquette/server/config/MemoryConfig.java index a1c8cdb3..41affe75 100644 --- a/broker/src/main/java/io/moquette/server/config/MemoryConfig.java +++ b/broker/src/main/java/io/moquette/server/config/MemoryConfig.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/main/java/io/moquette/server/config/ResourceLoaderConfig.java b/broker/src/main/java/io/moquette/server/config/ResourceLoaderConfig.java index 8a824389..5d64601e 100644 --- a/broker/src/main/java/io/moquette/server/config/ResourceLoaderConfig.java +++ b/broker/src/main/java/io/moquette/server/config/ResourceLoaderConfig.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/main/java/io/moquette/server/netty/MessageBuilder.java b/broker/src/main/java/io/moquette/server/netty/MessageBuilder.java index b7e90b80..dc1dc758 100644 --- a/broker/src/main/java/io/moquette/server/netty/MessageBuilder.java +++ b/broker/src/main/java/io/moquette/server/netty/MessageBuilder.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/main/java/io/moquette/server/netty/MoquetteIdleTimeoutHandler.java b/broker/src/main/java/io/moquette/server/netty/MoquetteIdleTimeoutHandler.java index 3e93a9e7..122f9b70 100644 --- a/broker/src/main/java/io/moquette/server/netty/MoquetteIdleTimeoutHandler.java +++ b/broker/src/main/java/io/moquette/server/netty/MoquetteIdleTimeoutHandler.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java b/broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java index 9e307019..67993d6d 100644 --- a/broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java +++ b/broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java b/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java index f1cf944a..c81bfcc9 100644 --- a/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java +++ b/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/main/java/io/moquette/server/netty/NettyUtils.java b/broker/src/main/java/io/moquette/server/netty/NettyUtils.java index 283d8db9..471ffd19 100644 --- a/broker/src/main/java/io/moquette/server/netty/NettyUtils.java +++ b/broker/src/main/java/io/moquette/server/netty/NettyUtils.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/main/java/io/moquette/server/netty/metrics/BytesMetrics.java b/broker/src/main/java/io/moquette/server/netty/metrics/BytesMetrics.java index 9a8f9e69..43f3ac16 100644 --- a/broker/src/main/java/io/moquette/server/netty/metrics/BytesMetrics.java +++ b/broker/src/main/java/io/moquette/server/netty/metrics/BytesMetrics.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/main/java/io/moquette/server/netty/metrics/BytesMetricsCollector.java b/broker/src/main/java/io/moquette/server/netty/metrics/BytesMetricsCollector.java index 45d61e3e..4c841129 100644 --- a/broker/src/main/java/io/moquette/server/netty/metrics/BytesMetricsCollector.java +++ b/broker/src/main/java/io/moquette/server/netty/metrics/BytesMetricsCollector.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/main/java/io/moquette/server/netty/metrics/BytesMetricsHandler.java b/broker/src/main/java/io/moquette/server/netty/metrics/BytesMetricsHandler.java index c5e346d6..d93c0e17 100644 --- a/broker/src/main/java/io/moquette/server/netty/metrics/BytesMetricsHandler.java +++ b/broker/src/main/java/io/moquette/server/netty/metrics/BytesMetricsHandler.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/main/java/io/moquette/server/netty/metrics/MQTTMessageLogger.java b/broker/src/main/java/io/moquette/server/netty/metrics/MQTTMessageLogger.java index c4e905d9..c4edf7e4 100644 --- a/broker/src/main/java/io/moquette/server/netty/metrics/MQTTMessageLogger.java +++ b/broker/src/main/java/io/moquette/server/netty/metrics/MQTTMessageLogger.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/main/java/io/moquette/server/netty/metrics/MessageMetrics.java b/broker/src/main/java/io/moquette/server/netty/metrics/MessageMetrics.java index 909784f1..44f8b5fd 100644 --- a/broker/src/main/java/io/moquette/server/netty/metrics/MessageMetrics.java +++ b/broker/src/main/java/io/moquette/server/netty/metrics/MessageMetrics.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/main/java/io/moquette/server/netty/metrics/MessageMetricsCollector.java b/broker/src/main/java/io/moquette/server/netty/metrics/MessageMetricsCollector.java index dfab4bbb..9952f4ef 100644 --- a/broker/src/main/java/io/moquette/server/netty/metrics/MessageMetricsCollector.java +++ b/broker/src/main/java/io/moquette/server/netty/metrics/MessageMetricsCollector.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/main/java/io/moquette/server/netty/metrics/MessageMetricsHandler.java b/broker/src/main/java/io/moquette/server/netty/metrics/MessageMetricsHandler.java index dc8f102a..c4894d12 100644 --- a/broker/src/main/java/io/moquette/server/netty/metrics/MessageMetricsHandler.java +++ b/broker/src/main/java/io/moquette/server/netty/metrics/MessageMetricsHandler.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/main/java/io/moquette/spi/ClientSession.java b/broker/src/main/java/io/moquette/spi/ClientSession.java index 698d1e62..63efdfe8 100644 --- a/broker/src/main/java/io/moquette/spi/ClientSession.java +++ b/broker/src/main/java/io/moquette/spi/ClientSession.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/main/java/io/moquette/spi/IMatchingCondition.java b/broker/src/main/java/io/moquette/spi/IMatchingCondition.java index 997a3be1..a8edf2a6 100644 --- a/broker/src/main/java/io/moquette/spi/IMatchingCondition.java +++ b/broker/src/main/java/io/moquette/spi/IMatchingCondition.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/main/java/io/moquette/spi/IMessagesStore.java b/broker/src/main/java/io/moquette/spi/IMessagesStore.java index 63ce2925..8ef8039f 100644 --- a/broker/src/main/java/io/moquette/spi/IMessagesStore.java +++ b/broker/src/main/java/io/moquette/spi/IMessagesStore.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/main/java/io/moquette/spi/IMessaging.java b/broker/src/main/java/io/moquette/spi/IMessaging.java index 223cd572..713a0381 100644 --- a/broker/src/main/java/io/moquette/spi/IMessaging.java +++ b/broker/src/main/java/io/moquette/spi/IMessaging.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/main/java/io/moquette/spi/ISessionsStore.java b/broker/src/main/java/io/moquette/spi/ISessionsStore.java index 8baa9d4e..00845313 100644 --- a/broker/src/main/java/io/moquette/spi/ISessionsStore.java +++ b/broker/src/main/java/io/moquette/spi/ISessionsStore.java @@ -1,213 +1,213 @@ -/* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() - * ------------------------------------------------------ - * 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 java.util.concurrent.BlockingQueue; -import io.moquette.spi.IMessagesStore.StoredMessage; -import io.moquette.spi.impl.subscriptions.Subscription; -import io.moquette.spi.impl.subscriptions.Topic; - -/** - * Store used to handle the persistence of the subscriptions tree. - * - * @author andrea - */ -public interface ISessionsStore { - - class ClientTopicCouple { - - public final Topic topicFilter; - public final String clientID; - - public ClientTopicCouple(String clientID, Topic 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(); - - void updateCleanStatus(String clientID, boolean cleanSession); - - /** - * 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 sessionID - * the client ID - */ - void wipeSubscriptions(String sessionID); - - /** - * Return all topic filters to recreate the subscription tree. - */ - List listAllSubscriptions(); - - /** - * @param couple - * the subscription descriptor. - * @return the subscription stored by clientID and topicFilter, if any else null; - */ - Subscription getSubscription(ClientTopicCouple couple); - - /* - * @return all subscriptions stored. - */ - List getSubscriptions(); - - /** - * @param clientID - * the session client ID. - * @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); - - /** - * Returns all the sessions - * - * @return - */ - Collection getAllSessions(); - - void inFlightAck(String clientID, int messageID); - - /** - * Save the binding messageID, clientID - guid - * - * @param clientID - * the client ID - * @param messageID - * the message ID - * @param guid - * the uuid of the message to mark as inflight. - */ - void inFlight(String clientID, int messageID, MessageGUID guid); - - /** - * 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 guids for retained messages for the session - * - * @param clientID - * the client ID owning the queue. - * @return the list of queue message UUIDs. - */ - BlockingQueue queue(String clientID); - - void dropQueue(String clientID); - - void moveInFlightToSecondPhaseAckWaiting(String clientID, int messageID); - - /** - * @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. - */ - MessageGUID secondPhaseAcknowledged(String clientID, int messageID); - - MessageGUID mapToGuid(String clientID, int messageID); - - StoredMessage getInflightMessage(String clientID, int messageID); - - /** - * Returns the number of inflight messages for the given client ID - * - * @param clientID - * @return - */ - int getInflightMessagesNo(String clientID); - - /** - * Returns the size of the session queue for the given client ID - * - * @param clientID - * @return - */ - int getPendingPublishMessagesNo(String clientID); - - /** - * Returns the number of second-phase ACK pending messages for the given client ID - * - * @param clientID - * @return - */ - int getSecondPhaseAckPendingMessages(String clientID); -} +/* + * Copyright (c) 2012-2017 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 java.util.concurrent.BlockingQueue; +import io.moquette.spi.IMessagesStore.StoredMessage; +import io.moquette.spi.impl.subscriptions.Subscription; +import io.moquette.spi.impl.subscriptions.Topic; + +/** + * Store used to handle the persistence of the subscriptions tree. + * + * @author andrea + */ +public interface ISessionsStore { + + class ClientTopicCouple { + + public final Topic topicFilter; + public final String clientID; + + public ClientTopicCouple(String clientID, Topic 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(); + + void updateCleanStatus(String clientID, boolean cleanSession); + + /** + * 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 sessionID + * the client ID + */ + void wipeSubscriptions(String sessionID); + + /** + * Return all topic filters to recreate the subscription tree. + */ + List listAllSubscriptions(); + + /** + * @param couple + * the subscription descriptor. + * @return the subscription stored by clientID and topicFilter, if any else null; + */ + Subscription getSubscription(ClientTopicCouple couple); + + /* + * @return all subscriptions stored. + */ + List getSubscriptions(); + + /** + * @param clientID + * the session client ID. + * @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); + + /** + * Returns all the sessions + * + * @return + */ + Collection getAllSessions(); + + void inFlightAck(String clientID, int messageID); + + /** + * Save the binding messageID, clientID - guid + * + * @param clientID + * the client ID + * @param messageID + * the message ID + * @param guid + * the uuid of the message to mark as inflight. + */ + void inFlight(String clientID, int messageID, MessageGUID guid); + + /** + * 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 guids for retained messages for the session + * + * @param clientID + * the client ID owning the queue. + * @return the list of queue message UUIDs. + */ + BlockingQueue queue(String clientID); + + void dropQueue(String clientID); + + void moveInFlightToSecondPhaseAckWaiting(String clientID, int messageID); + + /** + * @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. + */ + MessageGUID secondPhaseAcknowledged(String clientID, int messageID); + + MessageGUID mapToGuid(String clientID, int messageID); + + StoredMessage getInflightMessage(String clientID, int messageID); + + /** + * Returns the number of inflight messages for the given client ID + * + * @param clientID + * @return + */ + int getInflightMessagesNo(String clientID); + + /** + * Returns the size of the session queue for the given client ID + * + * @param clientID + * @return + */ + int getPendingPublishMessagesNo(String clientID); + + /** + * Returns the number of second-phase ACK pending messages for the given client ID + * + * @param clientID + * @return + */ + int getSecondPhaseAckPendingMessages(String clientID); +} diff --git a/broker/src/main/java/io/moquette/spi/MessageGUID.java b/broker/src/main/java/io/moquette/spi/MessageGUID.java index b933ec60..e2366c99 100644 --- a/broker/src/main/java/io/moquette/spi/MessageGUID.java +++ b/broker/src/main/java/io/moquette/spi/MessageGUID.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 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 7d2c1086..ac4f5ec8 100644 --- a/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java +++ b/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java @@ -1,260 +1,260 @@ -/* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() - * ------------------------------------------------------ - * 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.BrokerConstants; -import io.moquette.interception.InterceptHandler; -import io.moquette.interception.Interceptor; -import io.moquette.interception.messages.*; -import io.moquette.server.config.IConfig; -import io.moquette.spi.impl.subscriptions.Subscription; -import io.netty.handler.codec.mqtt.MqttConnectMessage; -import io.netty.handler.codec.mqtt.MqttPublishMessage; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.concurrent.CopyOnWriteArrayList; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import java.util.concurrent.TimeUnit; -import static io.moquette.logging.LoggingUtils.getInterceptorIds; - -/** - * An interceptor that execute the interception tasks asynchronously. - * - * @author Wagner Macedo - */ -final class BrokerInterceptor implements Interceptor { - - private static final Logger LOG = LoggerFactory.getLogger(BrokerInterceptor.class); - private final Map, List> handlers; - private final ExecutorService executor; - - private BrokerInterceptor(int poolSize, List handlers) { - LOG.info("Initializing broker interceptor. InterceptorIds = {}.", getInterceptorIds(handlers)); - this.handlers = new HashMap<>(); - for (Class messageType : InterceptHandler.ALL_MESSAGE_TYPES) { - this.handlers.put(messageType, new CopyOnWriteArrayList()); - } - for (InterceptHandler handler : handlers) { - this.addInterceptHandler(handler); - } - executor = Executors.newFixedThreadPool(poolSize); - } - - /** - * Configures a broker interceptor, with a thread pool of one thread. - * - * @param handlers - */ - public BrokerInterceptor(List handlers) { - this(1, handlers); - } - - /** - * Configures a broker interceptor using the pool size specified in the IConfig argument. - */ - public BrokerInterceptor(IConfig props, List handlers) { - this(Integer.parseInt(props.getProperty(BrokerConstants.BROKER_INTERCEPTOR_THREAD_POOL_SIZE, "1")), handlers); - } - - /** - * Shutdown graciously the executor service - */ - void stop() { - LOG.info("Shutting down interceptor thread pool..."); - executor.shutdown(); - try { - LOG.info("Waiting for thread pool tasks to terminate..."); - executor.awaitTermination(10L, TimeUnit.SECONDS); - } catch (InterruptedException e) { - } - if (!executor.isTerminated()) { - LOG.warn("Forcing shutdown of interceptor thread pool..."); - executor.shutdownNow(); - } - } - - @Override - public void notifyClientConnected(final MqttConnectMessage msg) { - for (final InterceptHandler handler : this.handlers.get(InterceptConnectMessage.class)) { - LOG.debug( - "Sending MQTT CONNECT message to interceptor. MqttClientId = {}, interceptorId = {}.", - msg.payload().clientIdentifier(), - handler.getID()); - executor.execute(new Runnable() { - - @Override - public void run() { - handler.onConnect(new InterceptConnectMessage(msg)); - } - }); - } - } - - @Override - public void notifyClientDisconnected(final String clientID, final String username) { - for (final InterceptHandler handler : this.handlers.get(InterceptDisconnectMessage.class)) { - LOG.debug( - "Notifying MQTT client disconnection to interceptor. MqttClientId = {}, username = {}, " - + "interceptorId = {}.", - clientID, - username, - handler.getID()); - executor.execute(new Runnable() { - - @Override - public void run() { - handler.onDisconnect(new InterceptDisconnectMessage(clientID, username)); - } - }); - } - } - - @Override - public void notifyClientConnectionLost(final String clientID, final String username) { - for (final InterceptHandler handler : this.handlers.get(InterceptConnectionLostMessage.class)) { - LOG.debug( - "Notifying unexpected MQTT client disconnection to interceptor. MqttClientId = {}, " - + "username = {}, interceptorId = {}.", - clientID, - username, - handler.getID()); - executor.execute(new Runnable() { - - @Override - public void run() { - handler.onConnectionLost(new InterceptConnectionLostMessage(clientID, username)); - } - }); - } - } - - @Override - public void notifyTopicPublished(final MqttPublishMessage msg, final String clientID, final String username) { - int messageId = msg.variableHeader().messageId(); - String topic = msg.variableHeader().topicName(); - for (final InterceptHandler handler : this.handlers.get(InterceptPublishMessage.class)) { - LOG.debug( - "Notifying MQTT PUBLISH message to interceptor. MqttClientId = {}, messageId = {}, topic = {}, " - + "interceptorId = {}.", - clientID, - messageId, - topic, - handler.getID()); - executor.execute(new Runnable() { - - @Override - public void run() { - handler.onPublish(new InterceptPublishMessage(msg, clientID, username)); - } - }); - } - } - - @Override - public void notifyTopicSubscribed(final Subscription sub, final String username) { - for (final InterceptHandler handler : this.handlers.get(InterceptSubscribeMessage.class)) { - LOG.debug( - "Notifying MQTT SUBSCRIBE message to interceptor. MqttClientId = {}, topicFilter = {}, " - + "interceptorId = {}.", - sub.getClientId(), - sub.getTopicFilter(), - handler.getID()); - executor.execute(new Runnable() { - - @Override - public void run() { - handler.onSubscribe(new InterceptSubscribeMessage(sub, username)); - } - }); - } - } - - @Override - public void notifyTopicUnsubscribed(final String topic, final String clientID, final String username) { - for (final InterceptHandler handler : this.handlers.get(InterceptUnsubscribeMessage.class)) { - LOG.debug( - "Notifying MQTT UNSUBSCRIBE message to interceptor. MqttClientId = {}, topic = {}, " - + "interceptorId = {}.", - clientID, - topic, - handler.getID()); - executor.execute(new Runnable() { - - @Override - public void run() { - handler.onUnsubscribe(new InterceptUnsubscribeMessage(topic, clientID, username)); - } - }); - } - } - - @Override - public void notifyMessageAcknowledged(final InterceptAcknowledgedMessage msg) { - for (final InterceptHandler handler : this.handlers.get(InterceptAcknowledgedMessage.class)) { - LOG.debug( - "Notifying MQTT ACK message to interceptor. MqttClientId = {}, messageId = {}, topic = {}, " - + "interceptorId = {}.", - msg.getMsg().getClientID(), - msg.getMsg().getMessageID(), - msg.getTopic(), - handler.getID()); - executor.execute(new Runnable() { - - @Override - public void run() { - handler.onMessageAcknowledged(msg); - } - }); - } - } - - @Override - public void addInterceptHandler(InterceptHandler interceptHandler) { - Class[] interceptedMessageTypes = getInterceptedMessageTypes(interceptHandler); - LOG.info( - "Adding MQTT message interceptor. InterceptorId = {}, handledMessageTypes = {}.", - interceptHandler.getID(), - interceptedMessageTypes); - for (Class interceptMessageType : interceptedMessageTypes) { - this.handlers.get(interceptMessageType).add(interceptHandler); - } - } - - @Override - public void removeInterceptHandler(InterceptHandler interceptHandler) { - Class[] interceptedMessageTypes = getInterceptedMessageTypes(interceptHandler); - LOG.info( - "Removing MQTT message interceptor. InterceptorId = {}, handledMessageTypes = {}.", - interceptHandler.getID(), - interceptedMessageTypes); - for (Class interceptMessageType : interceptedMessageTypes) { - this.handlers.get(interceptMessageType).remove(interceptHandler); - } - } - - private static Class[] getInterceptedMessageTypes(InterceptHandler interceptHandler) { - Class[] interceptedMessageTypes = interceptHandler.getInterceptedMessageTypes(); - if (interceptedMessageTypes == null) { - return InterceptHandler.ALL_MESSAGE_TYPES; - } - return interceptedMessageTypes; - } -} +/* + * Copyright (c) 2012-2017 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.BrokerConstants; +import io.moquette.interception.InterceptHandler; +import io.moquette.interception.Interceptor; +import io.moquette.interception.messages.*; +import io.moquette.server.config.IConfig; +import io.moquette.spi.impl.subscriptions.Subscription; +import io.netty.handler.codec.mqtt.MqttConnectMessage; +import io.netty.handler.codec.mqtt.MqttPublishMessage; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import static io.moquette.logging.LoggingUtils.getInterceptorIds; + +/** + * An interceptor that execute the interception tasks asynchronously. + * + * @author Wagner Macedo + */ +final class BrokerInterceptor implements Interceptor { + + private static final Logger LOG = LoggerFactory.getLogger(BrokerInterceptor.class); + private final Map, List> handlers; + private final ExecutorService executor; + + private BrokerInterceptor(int poolSize, List handlers) { + LOG.info("Initializing broker interceptor. InterceptorIds = {}.", getInterceptorIds(handlers)); + this.handlers = new HashMap<>(); + for (Class messageType : InterceptHandler.ALL_MESSAGE_TYPES) { + this.handlers.put(messageType, new CopyOnWriteArrayList()); + } + for (InterceptHandler handler : handlers) { + this.addInterceptHandler(handler); + } + executor = Executors.newFixedThreadPool(poolSize); + } + + /** + * Configures a broker interceptor, with a thread pool of one thread. + * + * @param handlers + */ + public BrokerInterceptor(List handlers) { + this(1, handlers); + } + + /** + * Configures a broker interceptor using the pool size specified in the IConfig argument. + */ + public BrokerInterceptor(IConfig props, List handlers) { + this(Integer.parseInt(props.getProperty(BrokerConstants.BROKER_INTERCEPTOR_THREAD_POOL_SIZE, "1")), handlers); + } + + /** + * Shutdown graciously the executor service + */ + void stop() { + LOG.info("Shutting down interceptor thread pool..."); + executor.shutdown(); + try { + LOG.info("Waiting for thread pool tasks to terminate..."); + executor.awaitTermination(10L, TimeUnit.SECONDS); + } catch (InterruptedException e) { + } + if (!executor.isTerminated()) { + LOG.warn("Forcing shutdown of interceptor thread pool..."); + executor.shutdownNow(); + } + } + + @Override + public void notifyClientConnected(final MqttConnectMessage msg) { + for (final InterceptHandler handler : this.handlers.get(InterceptConnectMessage.class)) { + LOG.debug( + "Sending MQTT CONNECT message to interceptor. MqttClientId = {}, interceptorId = {}.", + msg.payload().clientIdentifier(), + handler.getID()); + executor.execute(new Runnable() { + + @Override + public void run() { + handler.onConnect(new InterceptConnectMessage(msg)); + } + }); + } + } + + @Override + public void notifyClientDisconnected(final String clientID, final String username) { + for (final InterceptHandler handler : this.handlers.get(InterceptDisconnectMessage.class)) { + LOG.debug( + "Notifying MQTT client disconnection to interceptor. MqttClientId = {}, username = {}, " + + "interceptorId = {}.", + clientID, + username, + handler.getID()); + executor.execute(new Runnable() { + + @Override + public void run() { + handler.onDisconnect(new InterceptDisconnectMessage(clientID, username)); + } + }); + } + } + + @Override + public void notifyClientConnectionLost(final String clientID, final String username) { + for (final InterceptHandler handler : this.handlers.get(InterceptConnectionLostMessage.class)) { + LOG.debug( + "Notifying unexpected MQTT client disconnection to interceptor. MqttClientId = {}, " + + "username = {}, interceptorId = {}.", + clientID, + username, + handler.getID()); + executor.execute(new Runnable() { + + @Override + public void run() { + handler.onConnectionLost(new InterceptConnectionLostMessage(clientID, username)); + } + }); + } + } + + @Override + public void notifyTopicPublished(final MqttPublishMessage msg, final String clientID, final String username) { + int messageId = msg.variableHeader().messageId(); + String topic = msg.variableHeader().topicName(); + for (final InterceptHandler handler : this.handlers.get(InterceptPublishMessage.class)) { + LOG.debug( + "Notifying MQTT PUBLISH message to interceptor. MqttClientId = {}, messageId = {}, topic = {}, " + + "interceptorId = {}.", + clientID, + messageId, + topic, + handler.getID()); + executor.execute(new Runnable() { + + @Override + public void run() { + handler.onPublish(new InterceptPublishMessage(msg, clientID, username)); + } + }); + } + } + + @Override + public void notifyTopicSubscribed(final Subscription sub, final String username) { + for (final InterceptHandler handler : this.handlers.get(InterceptSubscribeMessage.class)) { + LOG.debug( + "Notifying MQTT SUBSCRIBE message to interceptor. MqttClientId = {}, topicFilter = {}, " + + "interceptorId = {}.", + sub.getClientId(), + sub.getTopicFilter(), + handler.getID()); + executor.execute(new Runnable() { + + @Override + public void run() { + handler.onSubscribe(new InterceptSubscribeMessage(sub, username)); + } + }); + } + } + + @Override + public void notifyTopicUnsubscribed(final String topic, final String clientID, final String username) { + for (final InterceptHandler handler : this.handlers.get(InterceptUnsubscribeMessage.class)) { + LOG.debug( + "Notifying MQTT UNSUBSCRIBE message to interceptor. MqttClientId = {}, topic = {}, " + + "interceptorId = {}.", + clientID, + topic, + handler.getID()); + executor.execute(new Runnable() { + + @Override + public void run() { + handler.onUnsubscribe(new InterceptUnsubscribeMessage(topic, clientID, username)); + } + }); + } + } + + @Override + public void notifyMessageAcknowledged(final InterceptAcknowledgedMessage msg) { + for (final InterceptHandler handler : this.handlers.get(InterceptAcknowledgedMessage.class)) { + LOG.debug( + "Notifying MQTT ACK message to interceptor. MqttClientId = {}, messageId = {}, topic = {}, " + + "interceptorId = {}.", + msg.getMsg().getClientID(), + msg.getMsg().getMessageID(), + msg.getTopic(), + handler.getID()); + executor.execute(new Runnable() { + + @Override + public void run() { + handler.onMessageAcknowledged(msg); + } + }); + } + } + + @Override + public void addInterceptHandler(InterceptHandler interceptHandler) { + Class[] interceptedMessageTypes = getInterceptedMessageTypes(interceptHandler); + LOG.info( + "Adding MQTT message interceptor. InterceptorId = {}, handledMessageTypes = {}.", + interceptHandler.getID(), + interceptedMessageTypes); + for (Class interceptMessageType : interceptedMessageTypes) { + this.handlers.get(interceptMessageType).add(interceptHandler); + } + } + + @Override + public void removeInterceptHandler(InterceptHandler interceptHandler) { + Class[] interceptedMessageTypes = getInterceptedMessageTypes(interceptHandler); + LOG.info( + "Removing MQTT message interceptor. InterceptorId = {}, handledMessageTypes = {}.", + interceptHandler.getID(), + interceptedMessageTypes); + for (Class interceptMessageType : interceptedMessageTypes) { + this.handlers.get(interceptMessageType).remove(interceptHandler); + } + } + + private static Class[] getInterceptedMessageTypes(InterceptHandler interceptHandler) { + Class[] interceptedMessageTypes = interceptHandler.getInterceptedMessageTypes(); + if (interceptedMessageTypes == null) { + return InterceptHandler.ALL_MESSAGE_TYPES; + } + return interceptedMessageTypes; + } +} diff --git a/broker/src/main/java/io/moquette/spi/impl/DebugUtils.java b/broker/src/main/java/io/moquette/spi/impl/DebugUtils.java index 2a029fa8..41a1009a 100644 --- a/broker/src/main/java/io/moquette/spi/impl/DebugUtils.java +++ b/broker/src/main/java/io/moquette/spi/impl/DebugUtils.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java index f2877ff2..468245ae 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java index b98135ff..279b0b94 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/main/java/io/moquette/spi/impl/security/ACLFileParser.java b/broker/src/main/java/io/moquette/spi/impl/security/ACLFileParser.java index 13648d42..839577d0 100644 --- a/broker/src/main/java/io/moquette/spi/impl/security/ACLFileParser.java +++ b/broker/src/main/java/io/moquette/spi/impl/security/ACLFileParser.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/main/java/io/moquette/spi/impl/security/Authorization.java b/broker/src/main/java/io/moquette/spi/impl/security/Authorization.java index 25e9f6b8..688aa7ef 100644 --- a/broker/src/main/java/io/moquette/spi/impl/security/Authorization.java +++ b/broker/src/main/java/io/moquette/spi/impl/security/Authorization.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/main/java/io/moquette/spi/impl/security/AuthorizationsCollector.java b/broker/src/main/java/io/moquette/spi/impl/security/AuthorizationsCollector.java index d894befa..82e75dcb 100644 --- a/broker/src/main/java/io/moquette/spi/impl/security/AuthorizationsCollector.java +++ b/broker/src/main/java/io/moquette/spi/impl/security/AuthorizationsCollector.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/main/java/io/moquette/spi/impl/security/DenyAllAuthorizator.java b/broker/src/main/java/io/moquette/spi/impl/security/DenyAllAuthorizator.java index d81638ec..e7c2a4e9 100644 --- a/broker/src/main/java/io/moquette/spi/impl/security/DenyAllAuthorizator.java +++ b/broker/src/main/java/io/moquette/spi/impl/security/DenyAllAuthorizator.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/main/java/io/moquette/spi/impl/security/FileAuthenticator.java b/broker/src/main/java/io/moquette/spi/impl/security/FileAuthenticator.java index b8e5fc82..3e5eb938 100644 --- a/broker/src/main/java/io/moquette/spi/impl/security/FileAuthenticator.java +++ b/broker/src/main/java/io/moquette/spi/impl/security/FileAuthenticator.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/main/java/io/moquette/spi/impl/security/PermitAllAuthorizator.java b/broker/src/main/java/io/moquette/spi/impl/security/PermitAllAuthorizator.java index c484d67b..fda989d5 100644 --- a/broker/src/main/java/io/moquette/spi/impl/security/PermitAllAuthorizator.java +++ b/broker/src/main/java/io/moquette/spi/impl/security/PermitAllAuthorizator.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/main/java/io/moquette/spi/impl/security/ResourceAuthenticator.java b/broker/src/main/java/io/moquette/spi/impl/security/ResourceAuthenticator.java index 02d8bdc3..4291f10b 100644 --- a/broker/src/main/java/io/moquette/spi/impl/security/ResourceAuthenticator.java +++ b/broker/src/main/java/io/moquette/spi/impl/security/ResourceAuthenticator.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/main/java/io/moquette/spi/impl/subscriptions/Subscription.java b/broker/src/main/java/io/moquette/spi/impl/subscriptions/Subscription.java index 37df5c3e..05837e50 100644 --- a/broker/src/main/java/io/moquette/spi/impl/subscriptions/Subscription.java +++ b/broker/src/main/java/io/moquette/spi/impl/subscriptions/Subscription.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionsStore.java b/broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionsStore.java index a005a9e6..ffb0690a 100644 --- a/broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionsStore.java +++ b/broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionsStore.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/main/java/io/moquette/spi/impl/subscriptions/Token.java b/broker/src/main/java/io/moquette/spi/impl/subscriptions/Token.java index 3efbd652..110e1ebf 100644 --- a/broker/src/main/java/io/moquette/spi/impl/subscriptions/Token.java +++ b/broker/src/main/java/io/moquette/spi/impl/subscriptions/Token.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/main/java/io/moquette/spi/impl/subscriptions/Topic.java b/broker/src/main/java/io/moquette/spi/impl/subscriptions/Topic.java index 5c0552e5..753af328 100644 --- a/broker/src/main/java/io/moquette/spi/impl/subscriptions/Topic.java +++ b/broker/src/main/java/io/moquette/spi/impl/subscriptions/Topic.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/main/java/io/moquette/spi/impl/subscriptions/TreeNode.java b/broker/src/main/java/io/moquette/spi/impl/subscriptions/TreeNode.java index 8d56629c..2e105987 100644 --- a/broker/src/main/java/io/moquette/spi/impl/subscriptions/TreeNode.java +++ b/broker/src/main/java/io/moquette/spi/impl/subscriptions/TreeNode.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java b/broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java index c5283da9..44b5a53c 100644 --- a/broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java +++ b/broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/main/java/io/moquette/spi/persistence/MapDBPersistentStore.java b/broker/src/main/java/io/moquette/spi/persistence/MapDBPersistentStore.java index 801a1244..e9c8fe77 100644 --- a/broker/src/main/java/io/moquette/spi/persistence/MapDBPersistentStore.java +++ b/broker/src/main/java/io/moquette/spi/persistence/MapDBPersistentStore.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 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 3976d157..b748eb7e 100644 --- a/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java +++ b/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java @@ -1,348 +1,348 @@ -/* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() - * ------------------------------------------------------ - * 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.StoredMessage; -import io.moquette.spi.ISessionsStore; -import io.moquette.spi.MessageGUID; -import io.moquette.spi.impl.Utils; -import io.moquette.spi.impl.subscriptions.Subscription; -import io.moquette.spi.impl.subscriptions.Topic; -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.BlockingQueue; -import java.util.concurrent.ConcurrentHashMap; -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->[MessageId -> guid] - private ConcurrentMap> m_secondPhaseStore; - - private final DB m_db; - private final MapDBMessagesStore m_messagesStore; - - MapDBSessionsStore(DB db, MapDBMessagesStore 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_secondPhaseStore = m_db.getHashMap("secondPhase"); - } - - @Override - public void addNewSubscription(Subscription newSubscription) { - LOG.info( - "Adding new subscription. CId= {}, topics = {}.", - newSubscription.getClientId(), - newSubscription.getTopicFilter()); - final String clientID = newSubscription.getClientId(); - m_db.getHashMap("subscriptions_" + clientID).put(newSubscription.getTopicFilter(), newSubscription); - - if (LOG.isTraceEnabled()) { - LOG.trace( - "The subscription has been added. CId= {}, topics = {}, clientSubscriptions = {}.", - newSubscription.getClientId(), - newSubscription.getTopicFilter(), - m_db.getHashMap("subscriptions_" + clientID)); - } - } - - @Override - public void removeSubscription(Topic topicFilter, String clientID) { - LOG.info("Removing subscription. CId= {}, topics = {}.", clientID, topicFilter); - if (!m_db.exists("subscriptions_" + clientID)) { - return; - } - m_db.getHashMap("subscriptions_" + clientID).remove(topicFilter); - if (LOG.isDebugEnabled()) { - LOG.debug( - "The subscription has been removed. CId= {}, topics = {}, clientSubscriptions = {}.", - clientID, - topicFilter, - m_db.getHashMap("subscriptions_" + clientID)); - } - } - - @Override - public void wipeSubscriptions(String clientID) { - LOG.info("Wiping subscriptions. CId= {}.", clientID); - m_db.delete("subscriptions_" + clientID); - if (LOG.isDebugEnabled()) { - LOG.debug( - "The subscriptions have been removed. CId= {}, clientSubscriptions = {}.", - clientID, - m_db.getHashMap("subscriptions_" + clientID)); - } - } - - @Override - public List listAllSubscriptions() { - LOG.info("Retrieving existing subscriptions..."); - final List allSubscriptions = new ArrayList<>(); - for (String clientID : m_persistentSessions.keySet()) { - ConcurrentMap clientSubscriptions = m_db.getHashMap("subscriptions_" + clientID); - for (Topic topicFilter : clientSubscriptions.keySet()) { - allSubscriptions.add(new ClientTopicCouple(clientID, topicFilter)); - } - } - if (LOG.isTraceEnabled()) { - LOG.trace("The existing subscriptions have been retrieved. Result = {}.", allSubscriptions); - } - return allSubscriptions; - } - - @Override - public Subscription getSubscription(ClientTopicCouple couple) { - ConcurrentMap clientSubscriptions = m_db.getHashMap("subscriptions_" + couple.clientID); - LOG.info("Retrieving subscriptions. CId= {}, subscriptions = {}.", couple.clientID, clientSubscriptions); - return clientSubscriptions.get(couple.topicFilter); - } - - @Override - public List getSubscriptions() { - LOG.info("Retrieving existing subscriptions..."); - List subscriptions = new ArrayList<>(); - for (String clientID : m_persistentSessions.keySet()) { - ConcurrentMap clientSubscriptions = m_db.getHashMap("subscriptions_" + clientID); - subscriptions.addAll(clientSubscriptions.values()); - } - LOG.debug("The existing subscriptions have been retrieved. Result = {}.", subscriptions); - return subscriptions; - } - - @Override - public boolean contains(String clientID) { - return m_db.exists("subscriptions_" + clientID); - } - - @Override - public ClientSession createNewSession(String clientID, boolean cleanSession) { - if (m_persistentSessions.containsKey(clientID)) { - LOG.error( - "Unable to create a new session: the client ID is already in use. CId= {}, cleanSession = {}.", - clientID, - cleanSession); - throw new IllegalArgumentException("Can't create a session with the ID of an already existing" + clientID); - } - LOG.info("Creating new session. CId= {}, cleanSession = {}.", clientID, cleanSession); - m_persistentSessions.putIfAbsent(clientID, new PersistentSession(cleanSession)); - return new ClientSession(clientID, m_messagesStore, this, cleanSession); - } - - @Override - public ClientSession sessionForClient(String clientID) { - LOG.info("Retrieving session. CId= {}.", clientID); - if (!m_persistentSessions.containsKey(clientID)) { - LOG.warn("The session does not exist. CId= {}.", clientID); - return null; - } - - PersistentSession storedSession = m_persistentSessions.get(clientID); - return new ClientSession(clientID, m_messagesStore, this, storedSession.cleanSession); - } - - @Override - public Collection getAllSessions() { - Collection result = new ArrayList<>(); - for (Map.Entry entry : m_persistentSessions.entrySet()) { - result.add(new ClientSession(entry.getKey(), m_messagesStore, this, entry.getValue().cleanSession)); - } - return result; - } - - @Override - public void updateCleanStatus(String clientID, boolean cleanSession) { - LOG.info("Updating cleanSession flag. CId= {}, cleanSession = {}.", clientID, 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) { - if (LOG.isDebugEnabled()) { - LOG.debug("Generating next packet ID. CId= {}.", clientID); - } - Set inFlightForClient = this.m_inFlightIds.get(clientID); - if (inFlightForClient == null) { - int nextPacketId = 1; - inFlightForClient = Collections.newSetFromMap(new ConcurrentHashMap()); - 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); - if (LOG.isDebugEnabled()) { - LOG.debug("The next packet ID has been generated. CId= {}, result = {}.", clientID, nextPacketId); - } - return nextPacketId; - } - - @Override - public void inFlightAck(String clientID, int messageID) { - LOG.debug("Acknowledging inflight message. CId= {}, messageId = {}.", clientID, messageID); - Map m = this.m_inflightStore.get(clientID); - if (m == null) { - LOG.warn("Unable to retrieve inflight message record. CId= {}, messageId = {}.", clientID, messageID); - 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, MessageGUID guid) { - LOG.debug("Storing inflight message. CId= {}, messageId = {}, guid = {}.", clientID, messageID, guid); - ConcurrentMap m = this.m_inflightStore.get(clientID); - if (m == null) { - m = new ConcurrentHashMap<>(); - } - m.put(messageID, guid); - LOG.info("storing inflight clientID <{}> messageID {} guid <{}>", clientID, messageID, guid); - this.m_inflightStore.put(clientID, m); - } - - @Override - public BlockingQueue queue(String clientID) { - LOG.info("Queuing pending message. CId= {}, guid = {}.", clientID); - return this.m_db.getQueue(clientID); - } - - @Override - public void dropQueue(String clientID) { - LOG.info("Removing pending messages. CId= {}.", clientID); - this.m_db.delete(clientID); - } - - @Override - public void moveInFlightToSecondPhaseAckWaiting(String clientID, int messageID) { - if (LOG.isDebugEnabled()) { - LOG.debug("Moving inflight message to 2nd phase ack state. CId= {}, messageID = {}.", clientID, messageID); - } - Map m = this.m_inflightStore.get(clientID); - if (m == null) { - LOG.warn("Unable to retrieve inflight message record. CId= {}, messageId = {}.", clientID, messageID); - return; - } - MessageGUID guid = m.remove(messageID); - - // remove from the ids store - Set inFlightForClient = this.m_inFlightIds.get(clientID); - if (inFlightForClient != null) { - inFlightForClient.remove(messageID); - } - - Map messageIDs = Utils - .defaultGet(m_secondPhaseStore, clientID, new HashMap()); - messageIDs.put(messageID, guid); - m_secondPhaseStore.put(clientID, messageIDs); - } - - @Override - public MessageGUID secondPhaseAcknowledged(String clientID, int messageID) { - if (LOG.isDebugEnabled()) { - LOG.debug("Processing second phase ACK CId={}, messageId={}", clientID, messageID); - } - Map messageIDs = Utils - .defaultGet(m_secondPhaseStore, clientID, new HashMap()); - MessageGUID guid = messageIDs.remove(messageID); - m_secondPhaseStore.put(clientID, messageIDs); - return guid; - } - - @Override - public MessageGUID mapToGuid(String clientID, int messageID) { - LOG.debug("Mapping message ID to GUID CId={}, messageId={}", clientID, messageID); - ConcurrentMap messageIdToGuid = m_db.getHashMap(messageId2GuidsMapName(clientID)); - MessageGUID result = messageIdToGuid.get(messageID); - LOG.debug("Message ID has been mapped to a GUID CId={}, messageId={}, guid={}", clientID, messageID, result); - return result; - } - - static String messageId2GuidsMapName(String clientID) { - return "guidsMapping_" + clientID; - } - - @Override - public StoredMessage getInflightMessage(String clientID, int messageID) { - LOG.info("Retrieving inflight message CId={}, messageId={}", clientID, messageID); - Map clientEntries = m_inflightStore.get(clientID); - if (clientEntries == null) { - LOG.warn("The client has no inflight messages CId={}, messageId={}", clientID, messageID); - return null; - } - MessageGUID guid = clientEntries.get(messageID); - if (guid == null) { - LOG.warn("The message ID does not have an associated GUID. CId= {}, messageId = {}.", clientID, messageID); - return null; - } - return m_messagesStore.getMessageByGuid(guid); - } - - @Override - public int getInflightMessagesNo(String clientID) { - if (!m_inflightStore.containsKey(clientID)) - return 0; - else - return m_inflightStore.get(clientID).size(); - } - - @Override - public int getPendingPublishMessagesNo(String clientID) { - return m_messagesStore.getPendingPublishMessages(clientID); - } - - @Override - public int getSecondPhaseAckPendingMessages(String clientID) { - if (!m_secondPhaseStore.containsKey(clientID)) - return 0; - else - return m_secondPhaseStore.get(clientID).size(); - } -} +/* + * Copyright (c) 2012-2017 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.StoredMessage; +import io.moquette.spi.ISessionsStore; +import io.moquette.spi.MessageGUID; +import io.moquette.spi.impl.Utils; +import io.moquette.spi.impl.subscriptions.Subscription; +import io.moquette.spi.impl.subscriptions.Topic; +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.BlockingQueue; +import java.util.concurrent.ConcurrentHashMap; +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->[MessageId -> guid] + private ConcurrentMap> m_secondPhaseStore; + + private final DB m_db; + private final MapDBMessagesStore m_messagesStore; + + MapDBSessionsStore(DB db, MapDBMessagesStore 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_secondPhaseStore = m_db.getHashMap("secondPhase"); + } + + @Override + public void addNewSubscription(Subscription newSubscription) { + LOG.info( + "Adding new subscription. CId= {}, topics = {}.", + newSubscription.getClientId(), + newSubscription.getTopicFilter()); + final String clientID = newSubscription.getClientId(); + m_db.getHashMap("subscriptions_" + clientID).put(newSubscription.getTopicFilter(), newSubscription); + + if (LOG.isTraceEnabled()) { + LOG.trace( + "The subscription has been added. CId= {}, topics = {}, clientSubscriptions = {}.", + newSubscription.getClientId(), + newSubscription.getTopicFilter(), + m_db.getHashMap("subscriptions_" + clientID)); + } + } + + @Override + public void removeSubscription(Topic topicFilter, String clientID) { + LOG.info("Removing subscription. CId= {}, topics = {}.", clientID, topicFilter); + if (!m_db.exists("subscriptions_" + clientID)) { + return; + } + m_db.getHashMap("subscriptions_" + clientID).remove(topicFilter); + if (LOG.isDebugEnabled()) { + LOG.debug( + "The subscription has been removed. CId= {}, topics = {}, clientSubscriptions = {}.", + clientID, + topicFilter, + m_db.getHashMap("subscriptions_" + clientID)); + } + } + + @Override + public void wipeSubscriptions(String clientID) { + LOG.info("Wiping subscriptions. CId= {}.", clientID); + m_db.delete("subscriptions_" + clientID); + if (LOG.isDebugEnabled()) { + LOG.debug( + "The subscriptions have been removed. CId= {}, clientSubscriptions = {}.", + clientID, + m_db.getHashMap("subscriptions_" + clientID)); + } + } + + @Override + public List listAllSubscriptions() { + LOG.info("Retrieving existing subscriptions..."); + final List allSubscriptions = new ArrayList<>(); + for (String clientID : m_persistentSessions.keySet()) { + ConcurrentMap clientSubscriptions = m_db.getHashMap("subscriptions_" + clientID); + for (Topic topicFilter : clientSubscriptions.keySet()) { + allSubscriptions.add(new ClientTopicCouple(clientID, topicFilter)); + } + } + if (LOG.isTraceEnabled()) { + LOG.trace("The existing subscriptions have been retrieved. Result = {}.", allSubscriptions); + } + return allSubscriptions; + } + + @Override + public Subscription getSubscription(ClientTopicCouple couple) { + ConcurrentMap clientSubscriptions = m_db.getHashMap("subscriptions_" + couple.clientID); + LOG.info("Retrieving subscriptions. CId= {}, subscriptions = {}.", couple.clientID, clientSubscriptions); + return clientSubscriptions.get(couple.topicFilter); + } + + @Override + public List getSubscriptions() { + LOG.info("Retrieving existing subscriptions..."); + List subscriptions = new ArrayList<>(); + for (String clientID : m_persistentSessions.keySet()) { + ConcurrentMap clientSubscriptions = m_db.getHashMap("subscriptions_" + clientID); + subscriptions.addAll(clientSubscriptions.values()); + } + LOG.debug("The existing subscriptions have been retrieved. Result = {}.", subscriptions); + return subscriptions; + } + + @Override + public boolean contains(String clientID) { + return m_db.exists("subscriptions_" + clientID); + } + + @Override + public ClientSession createNewSession(String clientID, boolean cleanSession) { + if (m_persistentSessions.containsKey(clientID)) { + LOG.error( + "Unable to create a new session: the client ID is already in use. CId= {}, cleanSession = {}.", + clientID, + cleanSession); + throw new IllegalArgumentException("Can't create a session with the ID of an already existing" + clientID); + } + LOG.info("Creating new session. CId= {}, cleanSession = {}.", clientID, cleanSession); + m_persistentSessions.putIfAbsent(clientID, new PersistentSession(cleanSession)); + return new ClientSession(clientID, m_messagesStore, this, cleanSession); + } + + @Override + public ClientSession sessionForClient(String clientID) { + LOG.info("Retrieving session. CId= {}.", clientID); + if (!m_persistentSessions.containsKey(clientID)) { + LOG.warn("The session does not exist. CId= {}.", clientID); + return null; + } + + PersistentSession storedSession = m_persistentSessions.get(clientID); + return new ClientSession(clientID, m_messagesStore, this, storedSession.cleanSession); + } + + @Override + public Collection getAllSessions() { + Collection result = new ArrayList<>(); + for (Map.Entry entry : m_persistentSessions.entrySet()) { + result.add(new ClientSession(entry.getKey(), m_messagesStore, this, entry.getValue().cleanSession)); + } + return result; + } + + @Override + public void updateCleanStatus(String clientID, boolean cleanSession) { + LOG.info("Updating cleanSession flag. CId= {}, cleanSession = {}.", clientID, 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) { + if (LOG.isDebugEnabled()) { + LOG.debug("Generating next packet ID. CId= {}.", clientID); + } + Set inFlightForClient = this.m_inFlightIds.get(clientID); + if (inFlightForClient == null) { + int nextPacketId = 1; + inFlightForClient = Collections.newSetFromMap(new ConcurrentHashMap()); + 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); + if (LOG.isDebugEnabled()) { + LOG.debug("The next packet ID has been generated. CId= {}, result = {}.", clientID, nextPacketId); + } + return nextPacketId; + } + + @Override + public void inFlightAck(String clientID, int messageID) { + LOG.debug("Acknowledging inflight message. CId= {}, messageId = {}.", clientID, messageID); + Map m = this.m_inflightStore.get(clientID); + if (m == null) { + LOG.warn("Unable to retrieve inflight message record. CId= {}, messageId = {}.", clientID, messageID); + 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, MessageGUID guid) { + LOG.debug("Storing inflight message. CId= {}, messageId = {}, guid = {}.", clientID, messageID, guid); + ConcurrentMap m = this.m_inflightStore.get(clientID); + if (m == null) { + m = new ConcurrentHashMap<>(); + } + m.put(messageID, guid); + LOG.info("storing inflight clientID <{}> messageID {} guid <{}>", clientID, messageID, guid); + this.m_inflightStore.put(clientID, m); + } + + @Override + public BlockingQueue queue(String clientID) { + LOG.info("Queuing pending message. CId= {}, guid = {}.", clientID); + return this.m_db.getQueue(clientID); + } + + @Override + public void dropQueue(String clientID) { + LOG.info("Removing pending messages. CId= {}.", clientID); + this.m_db.delete(clientID); + } + + @Override + public void moveInFlightToSecondPhaseAckWaiting(String clientID, int messageID) { + if (LOG.isDebugEnabled()) { + LOG.debug("Moving inflight message to 2nd phase ack state. CId= {}, messageID = {}.", clientID, messageID); + } + Map m = this.m_inflightStore.get(clientID); + if (m == null) { + LOG.warn("Unable to retrieve inflight message record. CId= {}, messageId = {}.", clientID, messageID); + return; + } + MessageGUID guid = m.remove(messageID); + + // remove from the ids store + Set inFlightForClient = this.m_inFlightIds.get(clientID); + if (inFlightForClient != null) { + inFlightForClient.remove(messageID); + } + + Map messageIDs = Utils + .defaultGet(m_secondPhaseStore, clientID, new HashMap()); + messageIDs.put(messageID, guid); + m_secondPhaseStore.put(clientID, messageIDs); + } + + @Override + public MessageGUID secondPhaseAcknowledged(String clientID, int messageID) { + if (LOG.isDebugEnabled()) { + LOG.debug("Processing second phase ACK CId={}, messageId={}", clientID, messageID); + } + Map messageIDs = Utils + .defaultGet(m_secondPhaseStore, clientID, new HashMap()); + MessageGUID guid = messageIDs.remove(messageID); + m_secondPhaseStore.put(clientID, messageIDs); + return guid; + } + + @Override + public MessageGUID mapToGuid(String clientID, int messageID) { + LOG.debug("Mapping message ID to GUID CId={}, messageId={}", clientID, messageID); + ConcurrentMap messageIdToGuid = m_db.getHashMap(messageId2GuidsMapName(clientID)); + MessageGUID result = messageIdToGuid.get(messageID); + LOG.debug("Message ID has been mapped to a GUID CId={}, messageId={}, guid={}", clientID, messageID, result); + return result; + } + + static String messageId2GuidsMapName(String clientID) { + return "guidsMapping_" + clientID; + } + + @Override + public StoredMessage getInflightMessage(String clientID, int messageID) { + LOG.info("Retrieving inflight message CId={}, messageId={}", clientID, messageID); + Map clientEntries = m_inflightStore.get(clientID); + if (clientEntries == null) { + LOG.warn("The client has no inflight messages CId={}, messageId={}", clientID, messageID); + return null; + } + MessageGUID guid = clientEntries.get(messageID); + if (guid == null) { + LOG.warn("The message ID does not have an associated GUID. CId= {}, messageId = {}.", clientID, messageID); + return null; + } + return m_messagesStore.getMessageByGuid(guid); + } + + @Override + public int getInflightMessagesNo(String clientID) { + if (!m_inflightStore.containsKey(clientID)) + return 0; + else + return m_inflightStore.get(clientID).size(); + } + + @Override + public int getPendingPublishMessagesNo(String clientID) { + return m_messagesStore.getPendingPublishMessages(clientID); + } + + @Override + public int getSecondPhaseAckPendingMessages(String clientID) { + if (!m_secondPhaseStore.containsKey(clientID)) + return 0; + else + return m_secondPhaseStore.get(clientID).size(); + } +} diff --git a/broker/src/main/java/io/moquette/spi/security/IAuthenticator.java b/broker/src/main/java/io/moquette/spi/security/IAuthenticator.java index 8df8a090..99a73e2a 100644 --- a/broker/src/main/java/io/moquette/spi/security/IAuthenticator.java +++ b/broker/src/main/java/io/moquette/spi/security/IAuthenticator.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/main/java/io/moquette/spi/security/IAuthorizator.java b/broker/src/main/java/io/moquette/spi/security/IAuthorizator.java index 050163bc..1aaaf5ad 100644 --- a/broker/src/main/java/io/moquette/spi/security/IAuthorizator.java +++ b/broker/src/main/java/io/moquette/spi/security/IAuthorizator.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/main/java/io/moquette/spi/security/ISslContextCreator.java b/broker/src/main/java/io/moquette/spi/security/ISslContextCreator.java index 0466a41d..7efcde73 100644 --- a/broker/src/main/java/io/moquette/spi/security/ISslContextCreator.java +++ b/broker/src/main/java/io/moquette/spi/security/ISslContextCreator.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/test/java/io/moquette/Utils.java b/broker/src/test/java/io/moquette/Utils.java index 08aedbad..cce6d78b 100644 --- a/broker/src/test/java/io/moquette/Utils.java +++ b/broker/src/test/java/io/moquette/Utils.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/test/java/io/moquette/server/ConfigurationClassLoaderTest.java b/broker/src/test/java/io/moquette/server/ConfigurationClassLoaderTest.java index 184c101b..f875d9b8 100644 --- a/broker/src/test/java/io/moquette/server/ConfigurationClassLoaderTest.java +++ b/broker/src/test/java/io/moquette/server/ConfigurationClassLoaderTest.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/test/java/io/moquette/server/IntegrationUtils.java b/broker/src/test/java/io/moquette/server/IntegrationUtils.java index 3fedd569..26d3c03f 100644 --- a/broker/src/test/java/io/moquette/server/IntegrationUtils.java +++ b/broker/src/test/java/io/moquette/server/IntegrationUtils.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/test/java/io/moquette/server/MQTTWebSocket.java b/broker/src/test/java/io/moquette/server/MQTTWebSocket.java index 06b89708..2866aa25 100644 --- a/broker/src/test/java/io/moquette/server/MQTTWebSocket.java +++ b/broker/src/test/java/io/moquette/server/MQTTWebSocket.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/test/java/io/moquette/server/ServerIntegrationEmbeddedPublishTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationEmbeddedPublishTest.java index a83e75d3..5689d9c5 100644 --- a/broker/src/test/java/io/moquette/server/ServerIntegrationEmbeddedPublishTest.java +++ b/broker/src/test/java/io/moquette/server/ServerIntegrationEmbeddedPublishTest.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/test/java/io/moquette/server/ServerIntegrationFuseTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationFuseTest.java index ed936b54..3bf83814 100644 --- a/broker/src/test/java/io/moquette/server/ServerIntegrationFuseTest.java +++ b/broker/src/test/java/io/moquette/server/ServerIntegrationFuseTest.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java index 0c0421fa..39d61ec7 100644 --- a/broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java +++ b/broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/test/java/io/moquette/server/ServerIntegrationQoSValidationTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationQoSValidationTest.java index 9a6fcbb6..0e83a65e 100644 --- a/broker/src/test/java/io/moquette/server/ServerIntegrationQoSValidationTest.java +++ b/broker/src/test/java/io/moquette/server/ServerIntegrationQoSValidationTest.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/test/java/io/moquette/server/ServerIntegrationSSLClientAuthTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationSSLClientAuthTest.java index 1bb19594..735e61dc 100644 --- a/broker/src/test/java/io/moquette/server/ServerIntegrationSSLClientAuthTest.java +++ b/broker/src/test/java/io/moquette/server/ServerIntegrationSSLClientAuthTest.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/test/java/io/moquette/server/ServerIntegrationSSLTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationSSLTest.java index 3abee018..6ef4303f 100644 --- a/broker/src/test/java/io/moquette/server/ServerIntegrationSSLTest.java +++ b/broker/src/test/java/io/moquette/server/ServerIntegrationSSLTest.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/test/java/io/moquette/server/ServerIntegrationWebSocketTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationWebSocketTest.java index b4ed8ac9..2946234c 100644 --- a/broker/src/test/java/io/moquette/server/ServerIntegrationWebSocketTest.java +++ b/broker/src/test/java/io/moquette/server/ServerIntegrationWebSocketTest.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/test/java/io/moquette/server/ServerLowlevelMessagesIntegrationTests.java b/broker/src/test/java/io/moquette/server/ServerLowlevelMessagesIntegrationTests.java index 3b3890a8..3a0431df 100644 --- a/broker/src/test/java/io/moquette/server/ServerLowlevelMessagesIntegrationTests.java +++ b/broker/src/test/java/io/moquette/server/ServerLowlevelMessagesIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/test/java/io/moquette/server/ServerRestartIntegrationTest.java b/broker/src/test/java/io/moquette/server/ServerRestartIntegrationTest.java index 2c640459..f8bbc583 100644 --- a/broker/src/test/java/io/moquette/server/ServerRestartIntegrationTest.java +++ b/broker/src/test/java/io/moquette/server/ServerRestartIntegrationTest.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/test/java/io/moquette/server/config/ConfigurationParserTest.java b/broker/src/test/java/io/moquette/server/config/ConfigurationParserTest.java index 515dea40..03f87066 100644 --- a/broker/src/test/java/io/moquette/server/config/ConfigurationParserTest.java +++ b/broker/src/test/java/io/moquette/server/config/ConfigurationParserTest.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/test/java/io/moquette/spec/v3_1_1/connection/ConnectionIT.java b/broker/src/test/java/io/moquette/spec/v3_1_1/connection/ConnectionIT.java index a81b32d0..dd7896d2 100644 --- a/broker/src/test/java/io/moquette/spec/v3_1_1/connection/ConnectionIT.java +++ b/broker/src/test/java/io/moquette/spec/v3_1_1/connection/ConnectionIT.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 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 d09cc135..fcd29fcb 100644 --- a/broker/src/test/java/io/moquette/spi/impl/BrokerInterceptorTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/BrokerInterceptorTest.java @@ -1,174 +1,174 @@ -/* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() - * ------------------------------------------------------ - * 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.messages.*; -import io.moquette.server.netty.MessageBuilder; -import io.moquette.spi.impl.subscriptions.Subscription; -import io.moquette.spi.impl.subscriptions.Topic; -import io.netty.handler.codec.mqtt.MqttQoS; -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.assertEquals; -import static org.mockito.Matchers.refEq; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.verify; - -/** - * @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 String getID() { - return "MockObserver"; - } - - @Override - public Class[] getInterceptedMessageTypes() { - return InterceptHandler.ALL_MESSAGE_TYPES; - } - - @Override - public void onConnect(InterceptConnectMessage msg) { - n.set(40); - } - - @Override - public void onDisconnect(InterceptDisconnectMessage msg) { - n.set(50); - } - - @Override - public void onConnectionLost(InterceptConnectionLostMessage msg) { - n.set(30); - } - - @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 onMessageAcknowledged(InterceptAcknowledgedMessage 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(MessageBuilder.connect().build()); - interval(); - assertEquals(40, n.get()); - } - - @Test - public void testNotifyClientDisconnected() throws Exception { - interceptor.notifyClientDisconnected("cli1234", "cli1234"); - interval(); - assertEquals(50, n.get()); - } - - @Test - public void testNotifyTopicPublished() throws Exception { - interceptor.notifyTopicPublished( - MessageBuilder.publish().qos(MqttQoS.AT_MOST_ONCE).payload("Hello".getBytes()).build(), - "cli1234", - "cli1234"); - interval(); - assertEquals(60, n.get()); - } - - @Test - public void testNotifyTopicSubscribed() throws Exception { - interceptor.notifyTopicSubscribed(new Subscription("cli1", new Topic("o2"), MqttQoS.AT_MOST_ONCE), "cli1234"); - interval(); - assertEquals(70, n.get()); - } - - @Test - public void testNotifyTopicUnsubscribed() throws Exception { - interceptor.notifyTopicUnsubscribed("o2", "cli1234", "cli1234"); - interval(); - assertEquals(80, n.get()); - } - - @Test - public void testAddAndRemoveInterceptHandler() throws Exception { - InterceptHandler interceptHandlerMock1 = mock(InterceptHandler.class); - InterceptHandler interceptHandlerMock2 = mock(InterceptHandler.class); - // add - interceptor.addInterceptHandler(interceptHandlerMock1); - interceptor.addInterceptHandler(interceptHandlerMock2); - - Subscription subscription = new Subscription("cli1", new Topic("o2"), MqttQoS.AT_MOST_ONCE); - interceptor.notifyTopicSubscribed(subscription, "cli1234"); - interval(); - - verify(interceptHandlerMock1).onSubscribe(refEq(new InterceptSubscribeMessage(subscription, "cli1234"))); - verify(interceptHandlerMock2).onSubscribe(refEq(new InterceptSubscribeMessage(subscription, "cli1234"))); - - // remove - interceptor.removeInterceptHandler(interceptHandlerMock1); - - interceptor.notifyTopicSubscribed(subscription, "cli1235"); - interval(); - // removeInterceptHandler() performs another interaction - // TODO: fix this - // verifyNoMoreInteractions(interceptHandlerMock1); - verify(interceptHandlerMock2).onSubscribe(refEq(new InterceptSubscribeMessage(subscription, "cli1235"))); - } -} +/* + * Copyright (c) 2012-2017 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.messages.*; +import io.moquette.server.netty.MessageBuilder; +import io.moquette.spi.impl.subscriptions.Subscription; +import io.moquette.spi.impl.subscriptions.Topic; +import io.netty.handler.codec.mqtt.MqttQoS; +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.assertEquals; +import static org.mockito.Matchers.refEq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; + +/** + * @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 String getID() { + return "MockObserver"; + } + + @Override + public Class[] getInterceptedMessageTypes() { + return InterceptHandler.ALL_MESSAGE_TYPES; + } + + @Override + public void onConnect(InterceptConnectMessage msg) { + n.set(40); + } + + @Override + public void onDisconnect(InterceptDisconnectMessage msg) { + n.set(50); + } + + @Override + public void onConnectionLost(InterceptConnectionLostMessage msg) { + n.set(30); + } + + @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 onMessageAcknowledged(InterceptAcknowledgedMessage 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(MessageBuilder.connect().build()); + interval(); + assertEquals(40, n.get()); + } + + @Test + public void testNotifyClientDisconnected() throws Exception { + interceptor.notifyClientDisconnected("cli1234", "cli1234"); + interval(); + assertEquals(50, n.get()); + } + + @Test + public void testNotifyTopicPublished() throws Exception { + interceptor.notifyTopicPublished( + MessageBuilder.publish().qos(MqttQoS.AT_MOST_ONCE).payload("Hello".getBytes()).build(), + "cli1234", + "cli1234"); + interval(); + assertEquals(60, n.get()); + } + + @Test + public void testNotifyTopicSubscribed() throws Exception { + interceptor.notifyTopicSubscribed(new Subscription("cli1", new Topic("o2"), MqttQoS.AT_MOST_ONCE), "cli1234"); + interval(); + assertEquals(70, n.get()); + } + + @Test + public void testNotifyTopicUnsubscribed() throws Exception { + interceptor.notifyTopicUnsubscribed("o2", "cli1234", "cli1234"); + interval(); + assertEquals(80, n.get()); + } + + @Test + public void testAddAndRemoveInterceptHandler() throws Exception { + InterceptHandler interceptHandlerMock1 = mock(InterceptHandler.class); + InterceptHandler interceptHandlerMock2 = mock(InterceptHandler.class); + // add + interceptor.addInterceptHandler(interceptHandlerMock1); + interceptor.addInterceptHandler(interceptHandlerMock2); + + Subscription subscription = new Subscription("cli1", new Topic("o2"), MqttQoS.AT_MOST_ONCE); + interceptor.notifyTopicSubscribed(subscription, "cli1234"); + interval(); + + verify(interceptHandlerMock1).onSubscribe(refEq(new InterceptSubscribeMessage(subscription, "cli1234"))); + verify(interceptHandlerMock2).onSubscribe(refEq(new InterceptSubscribeMessage(subscription, "cli1234"))); + + // remove + interceptor.removeInterceptHandler(interceptHandlerMock1); + + interceptor.notifyTopicSubscribed(subscription, "cli1235"); + interval(); + // removeInterceptHandler() performs another interaction + // TODO: fix this + // verifyNoMoreInteractions(interceptHandlerMock1); + verify(interceptHandlerMock2).onSubscribe(refEq(new InterceptSubscribeMessage(subscription, "cli1235"))); + } +} diff --git a/broker/src/test/java/io/moquette/spi/impl/MemoryMessagesStore.java b/broker/src/test/java/io/moquette/spi/impl/MemoryMessagesStore.java index 0e481866..2e838a96 100644 --- a/broker/src/test/java/io/moquette/spi/impl/MemoryMessagesStore.java +++ b/broker/src/test/java/io/moquette/spi/impl/MemoryMessagesStore.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/test/java/io/moquette/spi/impl/MemoryStorageService.java b/broker/src/test/java/io/moquette/spi/impl/MemoryStorageService.java index 78fcedf0..0d2627ca 100644 --- a/broker/src/test/java/io/moquette/spi/impl/MemoryStorageService.java +++ b/broker/src/test/java/io/moquette/spi/impl/MemoryStorageService.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/test/java/io/moquette/spi/impl/MockAuthenticator.java b/broker/src/test/java/io/moquette/spi/impl/MockAuthenticator.java index 5f054dab..f186c9fb 100644 --- a/broker/src/test/java/io/moquette/spi/impl/MockAuthenticator.java +++ b/broker/src/test/java/io/moquette/spi/impl/MockAuthenticator.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/test/java/io/moquette/spi/impl/MockReceiverChannel.java b/broker/src/test/java/io/moquette/spi/impl/MockReceiverChannel.java index fac2260c..1c3638c4 100644 --- a/broker/src/test/java/io/moquette/spi/impl/MockReceiverChannel.java +++ b/broker/src/test/java/io/moquette/spi/impl/MockReceiverChannel.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/test/java/io/moquette/spi/impl/NettyChannelAssertions.java b/broker/src/test/java/io/moquette/spi/impl/NettyChannelAssertions.java index 9d8daeab..67ae68e9 100644 --- a/broker/src/test/java/io/moquette/spi/impl/NettyChannelAssertions.java +++ b/broker/src/test/java/io/moquette/spi/impl/NettyChannelAssertions.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java index 580e92a4..d5bf38bb 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java index b50788fd..18e35e3f 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/test/java/io/moquette/spi/impl/security/ACLFileParserTest.java b/broker/src/test/java/io/moquette/spi/impl/security/ACLFileParserTest.java index a01190c3..7914916d 100644 --- a/broker/src/test/java/io/moquette/spi/impl/security/ACLFileParserTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/security/ACLFileParserTest.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/test/java/io/moquette/spi/impl/security/AuthorizationsCollectorTest.java b/broker/src/test/java/io/moquette/spi/impl/security/AuthorizationsCollectorTest.java index e2c87220..082b00a1 100644 --- a/broker/src/test/java/io/moquette/spi/impl/security/AuthorizationsCollectorTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/security/AuthorizationsCollectorTest.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/test/java/io/moquette/spi/impl/security/FileAuthenticatorTest.java b/broker/src/test/java/io/moquette/spi/impl/security/FileAuthenticatorTest.java index c539ff10..63bd2b74 100644 --- a/broker/src/test/java/io/moquette/spi/impl/security/FileAuthenticatorTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/security/FileAuthenticatorTest.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/test/java/io/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java b/broker/src/test/java/io/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java index 37ddb9f6..7d3ea50b 100644 --- a/broker/src/test/java/io/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/test/java/io/moquette/spi/impl/subscriptions/TopicTest.java b/broker/src/test/java/io/moquette/spi/impl/subscriptions/TopicTest.java index bfa53500..94bf2005 100644 --- a/broker/src/test/java/io/moquette/spi/impl/subscriptions/TopicTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/subscriptions/TopicTest.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/test/java/io/moquette/spi/persistence/MapDBPersistentStoreTest.java b/broker/src/test/java/io/moquette/spi/persistence/MapDBPersistentStoreTest.java index 96a4974b..cea1b6a3 100644 --- a/broker/src/test/java/io/moquette/spi/persistence/MapDBPersistentStoreTest.java +++ b/broker/src/test/java/io/moquette/spi/persistence/MapDBPersistentStoreTest.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 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 aa8a240e..3d6df059 100644 --- a/broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java +++ b/broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java @@ -1,273 +1,273 @@ -/* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() - * ------------------------------------------------------ - * 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.server.Constants; -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.MessageGUID; -import io.moquette.spi.impl.Utils; -import io.moquette.spi.impl.subscriptions.Subscription; -import io.moquette.spi.impl.subscriptions.Topic; -import io.moquette.spi.persistence.MapDBPersistentStore.PersistentSession; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import java.util.*; -import java.util.concurrent.ArrayBlockingQueue; -import java.util.concurrent.BlockingQueue; - -/** - * - * @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->BlockingQueue - private Map> queues = new HashMap<>(); - // maps clientID->[MessageId -> guid] - 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(Topic topic, String clientID) { - LOG.debug("removeSubscription topic filter: {} for clientID: {}", topic, clientID); - if (!m_persistentSubscriptions.containsKey(clientID)) { - return; - } - Map clientSubscriptions = m_persistentSubscriptions.get(clientID); - clientSubscriptions.remove(topic); - } - - @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 HashMap()); - } - - m_persistentSubscriptions.get(clientID).put(newSubscription.getTopicFilter(), newSubscription); - } - - @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 HashMap()); - 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 Collection getAllSessions() { - Collection result = new ArrayList<>(); - for (Map.Entry entry : m_persistentSessions.entrySet()) { - result.add(new ClientSession(entry.getKey(), m_messagesStore, this, entry.getValue().cleanSession)); - } - return result; - } - - @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().values()) { - allSubscriptions.add(sub.asClientTopicCouple()); - } - } - return allSubscriptions; - } - - @Override - public Subscription getSubscription(ClientTopicCouple couple) { - Map subscriptions = m_persistentSubscriptions.get(couple.clientID); - if (subscriptions == null || subscriptions.isEmpty()) { - return null; - } - return subscriptions.get(couple.topicFilter); - } - - @Override - public List getSubscriptions() { - List subscriptions = new ArrayList<>(); - for (Map.Entry> entry : m_persistentSubscriptions.entrySet()) { - subscriptions.addAll(entry.getValue().values()); - } - return subscriptions; - } - - @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, MessageGUID 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 BlockingQueue queue(String clientID) { - return Utils.defaultGet(queues, clientID, new ArrayBlockingQueue(Constants.MAX_MESSAGE_QUEUE)); - } - - @Override - public void dropQueue(String clientID) { - queues.remove(clientID); - } - - @Override - public void moveInFlightToSecondPhaseAckWaiting(String clientID, int messageID) { - LOG.info("acknowledging inflight clientID <{}> messageID {}", clientID, messageID); - Map m = this.m_inflightStore.get(clientID); - if (m == null) { - LOG.error("Can't find the inFlight record for client <{}>", clientID); - return; - } - MessageGUID guid = m.remove(messageID); - - LOG.info("Moving to second phase store"); - Map messageIDs = Utils - .defaultGet(m_secondPhaseStore, clientID, new HashMap()); - messageIDs.put(messageID, guid); - m_secondPhaseStore.put(clientID, messageIDs); - } - - @Override - public MessageGUID secondPhaseAcknowledged(String clientID, int messageID) { - Map messageIDs = Utils - .defaultGet(m_secondPhaseStore, clientID, new HashMap()); - MessageGUID guid = messageIDs.remove(messageID); - m_secondPhaseStore.put(clientID, messageIDs); - return guid; - } - - @Override - public MessageGUID 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; - } - - @Override - public int getInflightMessagesNo(String clientID) { - Map inflightMessages = m_inflightStore.get(clientID); - if (inflightMessages == null) - return 0; - else - return inflightMessages.size(); - } - - @Override - public int getPendingPublishMessagesNo(String clientID) { - return m_messagesStore.getPendingPublishMessages(clientID); - } - - @Override - public int getSecondPhaseAckPendingMessages(String clientID) { - Map pendingAcks = m_secondPhaseStore.get(clientID); - if (pendingAcks == null) - return 0; - else - return pendingAcks.size(); - } - -} +/* + * Copyright (c) 2012-2017 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.server.Constants; +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.MessageGUID; +import io.moquette.spi.impl.Utils; +import io.moquette.spi.impl.subscriptions.Subscription; +import io.moquette.spi.impl.subscriptions.Topic; +import io.moquette.spi.persistence.MapDBPersistentStore.PersistentSession; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import java.util.*; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.BlockingQueue; + +/** + * + * @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->BlockingQueue + private Map> queues = new HashMap<>(); + // maps clientID->[MessageId -> guid] + 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(Topic topic, String clientID) { + LOG.debug("removeSubscription topic filter: {} for clientID: {}", topic, clientID); + if (!m_persistentSubscriptions.containsKey(clientID)) { + return; + } + Map clientSubscriptions = m_persistentSubscriptions.get(clientID); + clientSubscriptions.remove(topic); + } + + @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 HashMap()); + } + + m_persistentSubscriptions.get(clientID).put(newSubscription.getTopicFilter(), newSubscription); + } + + @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 HashMap()); + 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 Collection getAllSessions() { + Collection result = new ArrayList<>(); + for (Map.Entry entry : m_persistentSessions.entrySet()) { + result.add(new ClientSession(entry.getKey(), m_messagesStore, this, entry.getValue().cleanSession)); + } + return result; + } + + @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().values()) { + allSubscriptions.add(sub.asClientTopicCouple()); + } + } + return allSubscriptions; + } + + @Override + public Subscription getSubscription(ClientTopicCouple couple) { + Map subscriptions = m_persistentSubscriptions.get(couple.clientID); + if (subscriptions == null || subscriptions.isEmpty()) { + return null; + } + return subscriptions.get(couple.topicFilter); + } + + @Override + public List getSubscriptions() { + List subscriptions = new ArrayList<>(); + for (Map.Entry> entry : m_persistentSubscriptions.entrySet()) { + subscriptions.addAll(entry.getValue().values()); + } + return subscriptions; + } + + @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, MessageGUID 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 BlockingQueue queue(String clientID) { + return Utils.defaultGet(queues, clientID, new ArrayBlockingQueue(Constants.MAX_MESSAGE_QUEUE)); + } + + @Override + public void dropQueue(String clientID) { + queues.remove(clientID); + } + + @Override + public void moveInFlightToSecondPhaseAckWaiting(String clientID, int messageID) { + LOG.info("acknowledging inflight clientID <{}> messageID {}", clientID, messageID); + Map m = this.m_inflightStore.get(clientID); + if (m == null) { + LOG.error("Can't find the inFlight record for client <{}>", clientID); + return; + } + MessageGUID guid = m.remove(messageID); + + LOG.info("Moving to second phase store"); + Map messageIDs = Utils + .defaultGet(m_secondPhaseStore, clientID, new HashMap()); + messageIDs.put(messageID, guid); + m_secondPhaseStore.put(clientID, messageIDs); + } + + @Override + public MessageGUID secondPhaseAcknowledged(String clientID, int messageID) { + Map messageIDs = Utils + .defaultGet(m_secondPhaseStore, clientID, new HashMap()); + MessageGUID guid = messageIDs.remove(messageID); + m_secondPhaseStore.put(clientID, messageIDs); + return guid; + } + + @Override + public MessageGUID 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; + } + + @Override + public int getInflightMessagesNo(String clientID) { + Map inflightMessages = m_inflightStore.get(clientID); + if (inflightMessages == null) + return 0; + else + return inflightMessages.size(); + } + + @Override + public int getPendingPublishMessagesNo(String clientID) { + return m_messagesStore.getPendingPublishMessages(clientID); + } + + @Override + public int getSecondPhaseAckPendingMessages(String clientID) { + Map pendingAcks = m_secondPhaseStore.get(clientID); + if (pendingAcks == null) + return 0; + else + return pendingAcks.size(); + } + +} diff --git a/broker/src/test/java/io/moquette/testclient/Client.java b/broker/src/test/java/io/moquette/testclient/Client.java index f8543cb0..e7115de3 100644 --- a/broker/src/test/java/io/moquette/testclient/Client.java +++ b/broker/src/test/java/io/moquette/testclient/Client.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/broker/src/test/java/io/moquette/testclient/ClientNettyMQTTHandler.java b/broker/src/test/java/io/moquette/testclient/ClientNettyMQTTHandler.java index 94073898..2a58ce07 100644 --- a/broker/src/test/java/io/moquette/testclient/ClientNettyMQTTHandler.java +++ b/broker/src/test/java/io/moquette/testclient/ClientNettyMQTTHandler.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 diff --git a/embedding_moquette/src/main/java/io/moquette/testembedded/EmbeddedLauncher.java b/embedding_moquette/src/main/java/io/moquette/testembedded/EmbeddedLauncher.java index 506b3ff8..dcbba09c 100644 --- a/embedding_moquette/src/main/java/io/moquette/testembedded/EmbeddedLauncher.java +++ b/embedding_moquette/src/main/java/io/moquette/testembedded/EmbeddedLauncher.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions() + * Copyright (c) 2012-2017 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 @@ -78,4 +78,4 @@ package io.moquette.testembedded; mqttBroker.internalPublish(message, "INTRLPUB"); System.out.println("After self publish"); } -} \ No newline at end of file +}