Minor fixed comment (#281)

2017-03-19 22:01:23 +01:00
parent 46d27e3bea
commit ec6ec09cee
87 changed files with 1536 additions and 1536 deletions
@@ -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
@@ -1,66 +1,66 @@
-/*
+/*
- * 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
+ * All rights reserved. This program and the accompanying materials
- * are made available under the terms of the Eclipse Public License v1.0
+ * are made available under the terms of the Eclipse Public License v1.0
- * and Apache License v2.0 which accompanies this distribution.
+ * and Apache License v2.0 which accompanies this distribution.
- *
+ *
- * The Eclipse Public License is available at
+ * The Eclipse Public License is available at
- * http://www.eclipse.org/legal/epl-v10.html
+ * http://www.eclipse.org/legal/epl-v10.html
- *
+ *
- * The Apache License v2.0 is available at
+ * The Apache License v2.0 is available at
- * http://www.opensource.org/licenses/apache2.0.php
+ * http://www.opensource.org/licenses/apache2.0.php
- *
+ *
- * You may elect to redistribute this code under either of these licenses.
+ * You may elect to redistribute this code under either of these licenses.
- */
+ */
-
+
-package io.moquette.interception;
+package io.moquette.interception;
-
+
-import io.moquette.interception.messages.InterceptAcknowledgedMessage;
+import io.moquette.interception.messages.InterceptAcknowledgedMessage;
-import io.moquette.interception.messages.InterceptConnectMessage;
+import io.moquette.interception.messages.InterceptConnectMessage;
-import io.moquette.interception.messages.InterceptConnectionLostMessage;
+import io.moquette.interception.messages.InterceptConnectionLostMessage;
-import io.moquette.interception.messages.InterceptDisconnectMessage;
+import io.moquette.interception.messages.InterceptDisconnectMessage;
-import io.moquette.interception.messages.InterceptPublishMessage;
+import io.moquette.interception.messages.InterceptPublishMessage;
-import io.moquette.interception.messages.InterceptSubscribeMessage;
+import io.moquette.interception.messages.InterceptSubscribeMessage;
-import io.moquette.interception.messages.InterceptUnsubscribeMessage;
+import io.moquette.interception.messages.InterceptUnsubscribeMessage;
-
+
-/**
+/**
- * Basic abstract class usefull to avoid empty methods creation in subclasses.
+ * Basic abstract class usefull to avoid empty methods creation in subclasses.
- *
+ *
- * Created by andrea on 08/12/15.
+ * Created by andrea on 08/12/15.
- */
+ */
-public abstract class AbstractInterceptHandler implements InterceptHandler {
+public abstract class AbstractInterceptHandler implements InterceptHandler {
-
+
-    @Override
+    @Override
-    public Class<?>[] getInterceptedMessageTypes() {
+    public Class<?>[] getInterceptedMessageTypes() {
-        return InterceptHandler.ALL_MESSAGE_TYPES;
+        return InterceptHandler.ALL_MESSAGE_TYPES;
-    }
+    }
-
+
-    @Override
+    @Override
-    public void onConnect(InterceptConnectMessage msg) {
+    public void onConnect(InterceptConnectMessage msg) {
-    }
+    }
-
+
-    @Override
+    @Override
-    public void onDisconnect(InterceptDisconnectMessage msg) {
+    public void onDisconnect(InterceptDisconnectMessage msg) {
-    }
+    }
-
+
-    @Override
+    @Override
-    public void onConnectionLost(InterceptConnectionLostMessage msg) {
+    public void onConnectionLost(InterceptConnectionLostMessage msg) {
-    }
+    }
-
+
-    @Override
+    @Override
-    public void onPublish(InterceptPublishMessage msg) {
+    public void onPublish(InterceptPublishMessage msg) {
-    }
+    }
-
+
-    @Override
+    @Override
-    public void onSubscribe(InterceptSubscribeMessage msg) {
+    public void onSubscribe(InterceptSubscribeMessage msg) {
-    }
+    }
-
+
-    @Override
+    @Override
-    public void onUnsubscribe(InterceptUnsubscribeMessage msg) {
+    public void onUnsubscribe(InterceptUnsubscribeMessage msg) {
-    }
+    }
-
+
-    @Override
+    @Override
-    public void onMessageAcknowledged(InterceptAcknowledgedMessage msg) {
+    public void onMessageAcknowledged(InterceptAcknowledgedMessage msg) {
-    }
+    }
-}
+}
@@ -1,68 +1,68 @@
-/*
+/*
- * 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
+ * All rights reserved. This program and the accompanying materials
- * are made available under the terms of the Eclipse Public License v1.0
+ * are made available under the terms of the Eclipse Public License v1.0
- * and Apache License v2.0 which accompanies this distribution.
+ * and Apache License v2.0 which accompanies this distribution.
- *
+ *
- * The Eclipse Public License is available at
+ * The Eclipse Public License is available at
- * http://www.eclipse.org/legal/epl-v10.html
+ * http://www.eclipse.org/legal/epl-v10.html
- *
+ *
- * The Apache License v2.0 is available at
+ * The Apache License v2.0 is available at
- * http://www.opensource.org/licenses/apache2.0.php
+ * http://www.opensource.org/licenses/apache2.0.php
- *
+ *
- * You may elect to redistribute this code under either of these licenses.
+ * You may elect to redistribute this code under either of these licenses.
- */
+ */
-
+
-package io.moquette.interception;
+package io.moquette.interception;
-
+
-import io.moquette.interception.messages.*;
+import io.moquette.interception.messages.*;
-import io.moquette.spi.impl.subscriptions.Subscription;
+import io.moquette.spi.impl.subscriptions.Subscription;
-import io.netty.handler.codec.mqtt.MqttMessage;
+import io.netty.handler.codec.mqtt.MqttMessage;
-
+
-/**
+/**
- * This interface is used to inject code for intercepting broker events.
+ * This interface is used to inject code for intercepting broker events.
- * <p>
+ * <p>
- * The events can act only as observers.
+ * The events can act only as observers.
- * <p>
+ * <p>
- * Almost every method receives a subclass of {@link MqttMessage}, except <code>onDisconnect</code>
+ * Almost every method receives a subclass of {@link MqttMessage}, except <code>onDisconnect</code>
- * that receives the client id string and <code>onSubscribe</code> and <code>onUnsubscribe</code>
+ * that receives the client id string and <code>onSubscribe</code> and <code>onUnsubscribe</code>
- * that receive a {@link Subscription} object.
+ * that receive a {@link Subscription} object.
- *
+ *
- * @author Wagner Macedo
+ * @author Wagner Macedo
- */
+ */
-public interface InterceptHandler {
+public interface InterceptHandler {
-
+
-    Class<?>[] ALL_MESSAGE_TYPES = {InterceptConnectMessage.class, InterceptDisconnectMessage.class,
+    Class<?>[] ALL_MESSAGE_TYPES = {InterceptConnectMessage.class, InterceptDisconnectMessage.class,
-            InterceptConnectionLostMessage.class, InterceptPublishMessage.class, InterceptSubscribeMessage.class,
+            InterceptConnectionLostMessage.class, InterceptPublishMessage.class, InterceptSubscribeMessage.class,
-            InterceptUnsubscribeMessage.class, InterceptAcknowledgedMessage.class};
+            InterceptUnsubscribeMessage.class, InterceptAcknowledgedMessage.class};
-
+
-    /**
+    /**
-     * Returns the identifier of this intercept handler.
+     * Returns the identifier of this intercept handler.
-     *
+     *
-     * @return
+     * @return
-     */
+     */
-    String getID();
+    String getID();
-
+
-    /**
+    /**
-     * Returns the InterceptMessage subtypes that this handler can process. If the result is null or
+     * 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.
+     * equal to ALL_MESSAGE_TYPES, all the message types will be processed.
-     *
+     *
-     * @return
+     * @return
-     */
+     */
-    Class<?>[] getInterceptedMessageTypes();
+    Class<?>[] getInterceptedMessageTypes();
-
+
-    void onConnect(InterceptConnectMessage msg);
+    void onConnect(InterceptConnectMessage msg);
-
+
-    void onDisconnect(InterceptDisconnectMessage msg);
+    void onDisconnect(InterceptDisconnectMessage msg);
-
+
-    void onConnectionLost(InterceptConnectionLostMessage msg);
+    void onConnectionLost(InterceptConnectionLostMessage msg);
-
+
-    void onPublish(InterceptPublishMessage msg);
+    void onPublish(InterceptPublishMessage msg);
-
+
-    void onSubscribe(InterceptSubscribeMessage msg);
+    void onSubscribe(InterceptSubscribeMessage msg);
-
+
-    void onUnsubscribe(InterceptUnsubscribeMessage msg);
+    void onUnsubscribe(InterceptUnsubscribeMessage msg);
-
+
-    void onMessageAcknowledged(InterceptAcknowledgedMessage msg);
+    void onMessageAcknowledged(InterceptAcknowledgedMessage msg);
-}
+}
@@ -1,54 +1,54 @@
-/*
+/*
- * 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
+ * All rights reserved. This program and the accompanying materials
- * are made available under the terms of the Eclipse Public License v1.0
+ * are made available under the terms of the Eclipse Public License v1.0
- * and Apache License v2.0 which accompanies this distribution.
+ * and Apache License v2.0 which accompanies this distribution.
- *
+ *
- * The Eclipse Public License is available at
+ * The Eclipse Public License is available at
- * http://www.eclipse.org/legal/epl-v10.html
+ * http://www.eclipse.org/legal/epl-v10.html
- *
+ *
- * The Apache License v2.0 is available at
+ * The Apache License v2.0 is available at
- * http://www.opensource.org/licenses/apache2.0.php
+ * http://www.opensource.org/licenses/apache2.0.php
- *
+ *
- * You may elect to redistribute this code under either of these licenses.
+ * You may elect to redistribute this code under either of these licenses.
- */
+ */
-
+
-package io.moquette.interception;
+package io.moquette.interception;
-
+
-import io.moquette.interception.messages.InterceptAcknowledgedMessage;
+import io.moquette.interception.messages.InterceptAcknowledgedMessage;
-import io.moquette.spi.impl.subscriptions.Subscription;
+import io.moquette.spi.impl.subscriptions.Subscription;
-import io.netty.handler.codec.mqtt.MqttConnectMessage;
+import io.netty.handler.codec.mqtt.MqttConnectMessage;
-import io.netty.handler.codec.mqtt.MqttPublishMessage;
+import io.netty.handler.codec.mqtt.MqttPublishMessage;
-
+
-/**
+/**
- * This interface is to be used internally by the broker components.
+ * This interface is to be used internally by the broker components.
- * <p>
+ * <p>
- * An interface is used instead of a class to allow more flexibility in changing an implementation.
+ * An interface is used instead of a class to allow more flexibility in changing an implementation.
- * <p>
+ * <p>
- * Interceptor implementations forward notifications to a <code>InterceptHandler</code>, that is
+ * Interceptor implementations forward notifications to a <code>InterceptHandler</code>, that is
- * normally a field. So, the implementations should act as a proxy to a custom intercept handler.
+ * normally a field. So, the implementations should act as a proxy to a custom intercept handler.
- *
+ *
- * @see InterceptHandler
+ * @see InterceptHandler
- * @author Wagner Macedo
+ * @author Wagner Macedo
- */
+ */
-public interface Interceptor {
+public interface Interceptor {
-
+
-    void notifyClientConnected(MqttConnectMessage msg);
+    void notifyClientConnected(MqttConnectMessage msg);
-
+
-    void notifyClientDisconnected(String clientID, String username);
+    void notifyClientDisconnected(String clientID, String username);
-
+
-    void notifyClientConnectionLost(String clientID, String username);
+    void notifyClientConnectionLost(String clientID, String username);
-
+
-    void notifyTopicPublished(MqttPublishMessage msg, String clientID, final String username);
+    void notifyTopicPublished(MqttPublishMessage msg, String clientID, final String username);
-
+
-    void notifyTopicSubscribed(Subscription sub, final String username);
+    void notifyTopicSubscribed(Subscription sub, final String username);
-
+
-    void notifyTopicUnsubscribed(String topic, String clientID, final String username);
+    void notifyTopicUnsubscribed(String topic, String clientID, final String username);
-
+
-    void notifyMessageAcknowledged(InterceptAcknowledgedMessage msg);
+    void notifyMessageAcknowledged(InterceptAcknowledgedMessage msg);
-
+
-    void addInterceptHandler(InterceptHandler interceptHandler);
+    void addInterceptHandler(InterceptHandler interceptHandler);
-
+
-    void removeInterceptHandler(InterceptHandler interceptHandler);
+    void removeInterceptHandler(InterceptHandler interceptHandler);
-}
+}
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -1,213 +1,213 @@
-/*
+/*
- * 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
+ * All rights reserved. This program and the accompanying materials
- * are made available under the terms of the Eclipse Public License v1.0
+ * are made available under the terms of the Eclipse Public License v1.0
- * and Apache License v2.0 which accompanies this distribution.
+ * and Apache License v2.0 which accompanies this distribution.
- *
+ *
- * The Eclipse Public License is available at
+ * The Eclipse Public License is available at
- * http://www.eclipse.org/legal/epl-v10.html
+ * http://www.eclipse.org/legal/epl-v10.html
- *
+ *
- * The Apache License v2.0 is available at
+ * The Apache License v2.0 is available at
- * http://www.opensource.org/licenses/apache2.0.php
+ * http://www.opensource.org/licenses/apache2.0.php
- *
+ *
- * You may elect to redistribute this code under either of these licenses.
+ * You may elect to redistribute this code under either of these licenses.
- */
+ */
-
+
-package io.moquette.spi;
+package io.moquette.spi;
-
+
-import java.util.Collection;
+import java.util.Collection;
-import java.util.List;
+import java.util.List;
-import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.BlockingQueue;
-import io.moquette.spi.IMessagesStore.StoredMessage;
+import io.moquette.spi.IMessagesStore.StoredMessage;
-import io.moquette.spi.impl.subscriptions.Subscription;
+import io.moquette.spi.impl.subscriptions.Subscription;
-import io.moquette.spi.impl.subscriptions.Topic;
+import io.moquette.spi.impl.subscriptions.Topic;
-
+
-/**
+/**
- * Store used to handle the persistence of the subscriptions tree.
+ * Store used to handle the persistence of the subscriptions tree.
- *
+ *
- * @author andrea
+ * @author andrea
- */
+ */
-public interface ISessionsStore {
+public interface ISessionsStore {
-
+
-    class ClientTopicCouple {
+    class ClientTopicCouple {
-
+
-        public final Topic topicFilter;
+        public final Topic topicFilter;
-        public final String clientID;
+        public final String clientID;
-
+
-        public ClientTopicCouple(String clientID, Topic topicFilter) {
+        public ClientTopicCouple(String clientID, Topic topicFilter) {
-            this.clientID = clientID;
+            this.clientID = clientID;
-            this.topicFilter = topicFilter;
+            this.topicFilter = topicFilter;
-        }
+        }
-
+
-        @Override
+        @Override
-        public boolean equals(Object o) {
+        public boolean equals(Object o) {
-            if (this == o)
+            if (this == o)
-                return true;
+                return true;
-            if (o == null || getClass() != o.getClass())
+            if (o == null || getClass() != o.getClass())
-                return false;
+                return false;
-
+
-            ClientTopicCouple that = (ClientTopicCouple) o;
+            ClientTopicCouple that = (ClientTopicCouple) o;
-
+
-            if (topicFilter != null ? !topicFilter.equals(that.topicFilter) : that.topicFilter != null)
+            if (topicFilter != null ? !topicFilter.equals(that.topicFilter) : that.topicFilter != null)
-                return false;
+                return false;
-            return !(clientID != null ? !clientID.equals(that.clientID) : that.clientID != null);
+            return !(clientID != null ? !clientID.equals(that.clientID) : that.clientID != null);
-        }
+        }
-
+
-        @Override
+        @Override
-        public int hashCode() {
+        public int hashCode() {
-            int result = topicFilter != null ? topicFilter.hashCode() : 0;
+            int result = topicFilter != null ? topicFilter.hashCode() : 0;
-            result = 31 * result + (clientID != null ? clientID.hashCode() : 0);
+            result = 31 * result + (clientID != null ? clientID.hashCode() : 0);
-            return result;
+            return result;
-        }
+        }
-
+
-        @Override
+        @Override
-        public String toString() {
+        public String toString() {
-            return "ClientTopicCouple{" + "topicFilter='" + topicFilter + '\'' + ", clientID='" + clientID + '\'' + '}';
+            return "ClientTopicCouple{" + "topicFilter='" + topicFilter + '\'' + ", clientID='" + clientID + '\'' + '}';
-        }
+        }
-
+
-    }
+    }
-
+
-    void initStore();
+    void initStore();
-
+
-    void updateCleanStatus(String clientID, boolean cleanSession);
+    void updateCleanStatus(String clientID, boolean cleanSession);
-
+
-    /**
+    /**
-     * Add a new subscription to the session
+     * Add a new subscription to the session
-     *
+     *
-     * @param newSubscription
+     * @param newSubscription
-     *            the subscription to add.
+     *            the subscription to add.
-     */
+     */
-    void addNewSubscription(Subscription newSubscription);
+    void addNewSubscription(Subscription newSubscription);
-
+
-    /**
+    /**
-     * Removed a specific subscription
+     * Removed a specific subscription
-     *
+     *
-     * @param topic
+     * @param topic
-     *            the topic of the subscription.
+     *            the topic of the subscription.
-     * @param clientID
+     * @param clientID
-     *            the session client.
+     *            the session client.
-     */
+     */
-    void removeSubscription(Topic topic, String clientID);
+    void removeSubscription(Topic topic, String clientID);
-
+
-    /**
+    /**
-     * Remove all the subscriptions of the session
+     * Remove all the subscriptions of the session
-     *
+     *
-     * @param sessionID
+     * @param sessionID
-     *            the client ID
+     *            the client ID
-     */
+     */
-    void wipeSubscriptions(String sessionID);
+    void wipeSubscriptions(String sessionID);
-
+
-    /**
+    /**
-     * Return all topic filters to recreate the subscription tree.
+     * Return all topic filters to recreate the subscription tree.
-     */
+     */
-    List<ClientTopicCouple> listAllSubscriptions();
+    List<ClientTopicCouple> listAllSubscriptions();
-
+
-    /**
+    /**
-     * @param couple
+     * @param couple
-     *            the subscription descriptor.
+     *            the subscription descriptor.
-     * @return the subscription stored by clientID and topicFilter, if any else null;
+     * @return the subscription stored by clientID and topicFilter, if any else null;
-     */
+     */
-    Subscription getSubscription(ClientTopicCouple couple);
+    Subscription getSubscription(ClientTopicCouple couple);
-
+
-    /*
+    /*
-     * @return all subscriptions stored.
+     * @return all subscriptions stored.
-     */
+     */
-    List<Subscription> getSubscriptions();
+    List<Subscription> getSubscriptions();
-
+
-    /**
+    /**
-     * @param clientID
+     * @param clientID
-     *            the session client ID.
+     *            the session client ID.
-     * @return true iff there are subscriptions persisted with clientID
+     * @return true iff there are subscriptions persisted with clientID
-     */
+     */
-    boolean contains(String clientID);
+    boolean contains(String clientID);
-
+
-    ClientSession createNewSession(String clientID, boolean cleanSession);
+    ClientSession createNewSession(String clientID, boolean cleanSession);
-
+
-    /**
+    /**
-     * @param clientID
+     * @param clientID
-     *            the client owning the session.
+     *            the client owning the session.
-     * @return the session for the given clientID, null if not found.
+     * @return the session for the given clientID, null if not found.
-     */
+     */
-    ClientSession sessionForClient(String clientID);
+    ClientSession sessionForClient(String clientID);
-
+
-    /**
+    /**
-     * Returns all the sessions
+     * Returns all the sessions
-     *
+     *
-     * @return
+     * @return
-     */
+     */
-    Collection<ClientSession> getAllSessions();
+    Collection<ClientSession> getAllSessions();
-
+
-    void inFlightAck(String clientID, int messageID);
+    void inFlightAck(String clientID, int messageID);
-
+
-    /**
+    /**
-     * Save the binding messageID, clientID - guid
+     * Save the binding messageID, clientID - guid
-     *
+     *
-     * @param clientID
+     * @param clientID
-     *            the client ID
+     *            the client ID
-     * @param messageID
+     * @param messageID
-     *            the message ID
+     *            the message ID
-     * @param guid
+     * @param guid
-     *            the uuid of the message to mark as inflight.
+     *            the uuid of the message to mark as inflight.
-     */
+     */
-    void inFlight(String clientID, int messageID, MessageGUID guid);
+    void inFlight(String clientID, int messageID, MessageGUID guid);
-
+
-    /**
+    /**
-     * Return the next valid packetIdentifier for the given client session.
+     * Return the next valid packetIdentifier for the given client session.
-     *
+     *
-     * @param clientID
+     * @param clientID
-     *            the clientID requesting next packet id.
+     *            the clientID requesting next packet id.
-     * @return the next valid id.
+     * @return the next valid id.
-     */
+     */
-    int nextPacketID(String clientID);
+    int nextPacketID(String clientID);
-
+
-    /**
+    /**
-     * List the guids for retained messages for the session
+     * List the guids for retained messages for the session
-     *
+     *
-     * @param clientID
+     * @param clientID
-     *            the client ID owning the queue.
+     *            the client ID owning the queue.
-     * @return the list of queue message UUIDs.
+     * @return the list of queue message UUIDs.
-     */
+     */
-    BlockingQueue<StoredMessage> queue(String clientID);
+    BlockingQueue<StoredMessage> queue(String clientID);
-
+
-    void dropQueue(String clientID);
+    void dropQueue(String clientID);
-
+
-    void moveInFlightToSecondPhaseAckWaiting(String clientID, int messageID);
+    void moveInFlightToSecondPhaseAckWaiting(String clientID, int messageID);
-
+
-    /**
+    /**
-     * @param clientID
+     * @param clientID
-     *            the client ID accessing the second phase.
+     *            the client ID accessing the second phase.
-     * @param messageID
+     * @param messageID
-     *            the message ID that reached the second phase.
+     *            the message ID that reached the second phase.
-     * @return the guid of message just acked.
+     * @return the guid of message just acked.
-     */
+     */
-    MessageGUID secondPhaseAcknowledged(String clientID, int messageID);
+    MessageGUID secondPhaseAcknowledged(String clientID, int messageID);
-
+
-    MessageGUID mapToGuid(String clientID, int messageID);
+    MessageGUID mapToGuid(String clientID, int messageID);
-
+
-    StoredMessage getInflightMessage(String clientID, int messageID);
+    StoredMessage getInflightMessage(String clientID, int messageID);
-
+
-    /**
+    /**
-     * Returns the number of inflight messages for the given client ID
+     * Returns the number of inflight messages for the given client ID
-     *
+     *
-     * @param clientID
+     * @param clientID
-     * @return
+     * @return
-     */
+     */
-    int getInflightMessagesNo(String clientID);
+    int getInflightMessagesNo(String clientID);
-
+
-    /**
+    /**
-     * Returns the size of the session queue for the given client ID
+     * Returns the size of the session queue for the given client ID
-     *
+     *
-     * @param clientID
+     * @param clientID
-     * @return
+     * @return
-     */
+     */
-    int getPendingPublishMessagesNo(String clientID);
+    int getPendingPublishMessagesNo(String clientID);
-
+
-    /**
+    /**
-     * Returns the number of second-phase ACK pending messages for the given client ID
+     * Returns the number of second-phase ACK pending messages for the given client ID
-     *
+     *
-     * @param clientID
+     * @param clientID
-     * @return
+     * @return
-     */
+     */
-    int getSecondPhaseAckPendingMessages(String clientID);
+    int getSecondPhaseAckPendingMessages(String clientID);
-}
+}
@@ -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
@@ -1,260 +1,260 @@
-/*
+/*
- * 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
+ * All rights reserved. This program and the accompanying materials
- * are made available under the terms of the Eclipse Public License v1.0
+ * are made available under the terms of the Eclipse Public License v1.0
- * and Apache License v2.0 which accompanies this distribution.
+ * and Apache License v2.0 which accompanies this distribution.
- *
+ *
- * The Eclipse Public License is available at
+ * The Eclipse Public License is available at
- * http://www.eclipse.org/legal/epl-v10.html
+ * http://www.eclipse.org/legal/epl-v10.html
- *
+ *
- * The Apache License v2.0 is available at
+ * The Apache License v2.0 is available at
- * http://www.opensource.org/licenses/apache2.0.php
+ * http://www.opensource.org/licenses/apache2.0.php
- *
+ *
- * You may elect to redistribute this code under either of these licenses.
+ * You may elect to redistribute this code under either of these licenses.
- */
+ */
-
+
-package io.moquette.spi.impl;
+package io.moquette.spi.impl;
-
+
-import io.moquette.BrokerConstants;
+import io.moquette.BrokerConstants;
-import io.moquette.interception.InterceptHandler;
+import io.moquette.interception.InterceptHandler;
-import io.moquette.interception.Interceptor;
+import io.moquette.interception.Interceptor;
-import io.moquette.interception.messages.*;
+import io.moquette.interception.messages.*;
-import io.moquette.server.config.IConfig;
+import io.moquette.server.config.IConfig;
-import io.moquette.spi.impl.subscriptions.Subscription;
+import io.moquette.spi.impl.subscriptions.Subscription;
-import io.netty.handler.codec.mqtt.MqttConnectMessage;
+import io.netty.handler.codec.mqtt.MqttConnectMessage;
-import io.netty.handler.codec.mqtt.MqttPublishMessage;
+import io.netty.handler.codec.mqtt.MqttPublishMessage;
-import org.slf4j.Logger;
+import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.slf4j.LoggerFactory;
-import java.util.HashMap;
+import java.util.HashMap;
-import java.util.List;
+import java.util.List;
-import java.util.Map;
+import java.util.Map;
-import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CopyOnWriteArrayList;
-import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
+import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeUnit;
-import static io.moquette.logging.LoggingUtils.getInterceptorIds;
+import static io.moquette.logging.LoggingUtils.getInterceptorIds;
-
+
-/**
+/**
- * An interceptor that execute the interception tasks asynchronously.
+ * An interceptor that execute the interception tasks asynchronously.
- *
+ *
- * @author Wagner Macedo
+ * @author Wagner Macedo
- */
+ */
-final class BrokerInterceptor implements Interceptor {
+final class BrokerInterceptor implements Interceptor {
-
+
-    private static final Logger LOG = LoggerFactory.getLogger(BrokerInterceptor.class);
+    private static final Logger LOG = LoggerFactory.getLogger(BrokerInterceptor.class);
-    private final Map<Class<?>, List<InterceptHandler>> handlers;
+    private final Map<Class<?>, List<InterceptHandler>> handlers;
-    private final ExecutorService executor;
+    private final ExecutorService executor;
-
+
-    private BrokerInterceptor(int poolSize, List<InterceptHandler> handlers) {
+    private BrokerInterceptor(int poolSize, List<InterceptHandler> handlers) {
-        LOG.info("Initializing broker interceptor. InterceptorIds = {}.", getInterceptorIds(handlers));
+        LOG.info("Initializing broker interceptor. InterceptorIds = {}.", getInterceptorIds(handlers));
-        this.handlers = new HashMap<>();
+        this.handlers = new HashMap<>();
-        for (Class<?> messageType : InterceptHandler.ALL_MESSAGE_TYPES) {
+        for (Class<?> messageType : InterceptHandler.ALL_MESSAGE_TYPES) {
-            this.handlers.put(messageType, new CopyOnWriteArrayList<InterceptHandler>());
+            this.handlers.put(messageType, new CopyOnWriteArrayList<InterceptHandler>());
-        }
+        }
-        for (InterceptHandler handler : handlers) {
+        for (InterceptHandler handler : handlers) {
-            this.addInterceptHandler(handler);
+            this.addInterceptHandler(handler);
-        }
+        }
-        executor = Executors.newFixedThreadPool(poolSize);
+        executor = Executors.newFixedThreadPool(poolSize);
-    }
+    }
-
+
-    /**
+    /**
-     * Configures a broker interceptor, with a thread pool of one thread.
+     * Configures a broker interceptor, with a thread pool of one thread.
-     *
+     *
-     * @param handlers
+     * @param handlers
-     */
+     */
-    public BrokerInterceptor(List<InterceptHandler> handlers) {
+    public BrokerInterceptor(List<InterceptHandler> handlers) {
-        this(1, handlers);
+        this(1, handlers);
-    }
+    }
-
+
-    /**
+    /**
-     * Configures a broker interceptor using the pool size specified in the IConfig argument.
+     * Configures a broker interceptor using the pool size specified in the IConfig argument.
-     */
+     */
-    public BrokerInterceptor(IConfig props, List<InterceptHandler> handlers) {
+    public BrokerInterceptor(IConfig props, List<InterceptHandler> handlers) {
-        this(Integer.parseInt(props.getProperty(BrokerConstants.BROKER_INTERCEPTOR_THREAD_POOL_SIZE, "1")), handlers);
+        this(Integer.parseInt(props.getProperty(BrokerConstants.BROKER_INTERCEPTOR_THREAD_POOL_SIZE, "1")), handlers);
-    }
+    }
-
+
-    /**
+    /**
-     * Shutdown graciously the executor service
+     * Shutdown graciously the executor service
-     */
+     */
-    void stop() {
+    void stop() {
-        LOG.info("Shutting down interceptor thread pool...");
+        LOG.info("Shutting down interceptor thread pool...");
-        executor.shutdown();
+        executor.shutdown();
-        try {
+        try {
-            LOG.info("Waiting for thread pool tasks to terminate...");
+            LOG.info("Waiting for thread pool tasks to terminate...");
-            executor.awaitTermination(10L, TimeUnit.SECONDS);
+            executor.awaitTermination(10L, TimeUnit.SECONDS);
-        } catch (InterruptedException e) {
+        } catch (InterruptedException e) {
-        }
+        }
-        if (!executor.isTerminated()) {
+        if (!executor.isTerminated()) {
-            LOG.warn("Forcing shutdown of interceptor thread pool...");
+            LOG.warn("Forcing shutdown of interceptor thread pool...");
-            executor.shutdownNow();
+            executor.shutdownNow();
-        }
+        }
-    }
+    }
-
+
-    @Override
+    @Override
-    public void notifyClientConnected(final MqttConnectMessage msg) {
+    public void notifyClientConnected(final MqttConnectMessage msg) {
-        for (final InterceptHandler handler : this.handlers.get(InterceptConnectMessage.class)) {
+        for (final InterceptHandler handler : this.handlers.get(InterceptConnectMessage.class)) {
-            LOG.debug(
+            LOG.debug(
-                    "Sending MQTT CONNECT message to interceptor. MqttClientId = {}, interceptorId = {}.",
+                    "Sending MQTT CONNECT message to interceptor. MqttClientId = {}, interceptorId = {}.",
-                    msg.payload().clientIdentifier(),
+                    msg.payload().clientIdentifier(),
-                    handler.getID());
+                    handler.getID());
-            executor.execute(new Runnable() {
+            executor.execute(new Runnable() {
-
+
-                @Override
+                @Override
-                public void run() {
+                public void run() {
-                    handler.onConnect(new InterceptConnectMessage(msg));
+                    handler.onConnect(new InterceptConnectMessage(msg));
-                }
+                }
-            });
+            });
-        }
+        }
-    }
+    }
-
+
-    @Override
+    @Override
-    public void notifyClientDisconnected(final String clientID, final String username) {
+    public void notifyClientDisconnected(final String clientID, final String username) {
-        for (final InterceptHandler handler : this.handlers.get(InterceptDisconnectMessage.class)) {
+        for (final InterceptHandler handler : this.handlers.get(InterceptDisconnectMessage.class)) {
-            LOG.debug(
+            LOG.debug(
-                    "Notifying MQTT client disconnection to interceptor. MqttClientId = {}, username = {}, "
+                    "Notifying MQTT client disconnection to interceptor. MqttClientId = {}, username = {}, "
-                            + "interceptorId = {}.",
+                            + "interceptorId = {}.",
-                    clientID,
+                    clientID,
-                    username,
+                    username,
-                    handler.getID());
+                    handler.getID());
-            executor.execute(new Runnable() {
+            executor.execute(new Runnable() {
-
+
-                @Override
+                @Override
-                public void run() {
+                public void run() {
-                    handler.onDisconnect(new InterceptDisconnectMessage(clientID, username));
+                    handler.onDisconnect(new InterceptDisconnectMessage(clientID, username));
-                }
+                }
-            });
+            });
-        }
+        }
-    }
+    }
-
+
-    @Override
+    @Override
-    public void notifyClientConnectionLost(final String clientID, final String username) {
+    public void notifyClientConnectionLost(final String clientID, final String username) {
-        for (final InterceptHandler handler : this.handlers.get(InterceptConnectionLostMessage.class)) {
+        for (final InterceptHandler handler : this.handlers.get(InterceptConnectionLostMessage.class)) {
-            LOG.debug(
+            LOG.debug(
-                    "Notifying unexpected MQTT client disconnection to interceptor. MqttClientId = {}, "
+                    "Notifying unexpected MQTT client disconnection to interceptor. MqttClientId = {}, "
-                            + "username = {}, interceptorId = {}.",
+                            + "username = {}, interceptorId = {}.",
-                    clientID,
+                    clientID,
-                    username,
+                    username,
-                    handler.getID());
+                    handler.getID());
-            executor.execute(new Runnable() {
+            executor.execute(new Runnable() {
-
+
-                @Override
+                @Override
-                public void run() {
+                public void run() {
-                    handler.onConnectionLost(new InterceptConnectionLostMessage(clientID, username));
+                    handler.onConnectionLost(new InterceptConnectionLostMessage(clientID, username));
-                }
+                }
-            });
+            });
-        }
+        }
-    }
+    }
-
+
-    @Override
+    @Override
-    public void notifyTopicPublished(final MqttPublishMessage msg, final String clientID, final String username) {
+    public void notifyTopicPublished(final MqttPublishMessage msg, final String clientID, final String username) {
-        int messageId = msg.variableHeader().messageId();
+        int messageId = msg.variableHeader().messageId();
-        String topic = msg.variableHeader().topicName();
+        String topic = msg.variableHeader().topicName();
-        for (final InterceptHandler handler : this.handlers.get(InterceptPublishMessage.class)) {
+        for (final InterceptHandler handler : this.handlers.get(InterceptPublishMessage.class)) {
-            LOG.debug(
+            LOG.debug(
-                    "Notifying MQTT PUBLISH message to interceptor. MqttClientId = {}, messageId = {}, topic = {}, "
+                    "Notifying MQTT PUBLISH message to interceptor. MqttClientId = {}, messageId = {}, topic = {}, "
-                            + "interceptorId = {}.",
+                            + "interceptorId = {}.",
-                    clientID,
+                    clientID,
-                    messageId,
+                    messageId,
-                    topic,
+                    topic,
-                    handler.getID());
+                    handler.getID());
-            executor.execute(new Runnable() {
+            executor.execute(new Runnable() {
-
+
-                @Override
+                @Override
-                public void run() {
+                public void run() {
-                    handler.onPublish(new InterceptPublishMessage(msg, clientID, username));
+                    handler.onPublish(new InterceptPublishMessage(msg, clientID, username));
-                }
+                }
-            });
+            });
-        }
+        }
-    }
+    }
-
+
-    @Override
+    @Override
-    public void notifyTopicSubscribed(final Subscription sub, final String username) {
+    public void notifyTopicSubscribed(final Subscription sub, final String username) {
-        for (final InterceptHandler handler : this.handlers.get(InterceptSubscribeMessage.class)) {
+        for (final InterceptHandler handler : this.handlers.get(InterceptSubscribeMessage.class)) {
-            LOG.debug(
+            LOG.debug(
-                    "Notifying MQTT SUBSCRIBE message to interceptor. MqttClientId = {}, topicFilter = {}, "
+                    "Notifying MQTT SUBSCRIBE message to interceptor. MqttClientId = {}, topicFilter = {}, "
-                            + "interceptorId = {}.",
+                            + "interceptorId = {}.",
-                    sub.getClientId(),
+                    sub.getClientId(),
-                    sub.getTopicFilter(),
+                    sub.getTopicFilter(),
-                    handler.getID());
+                    handler.getID());
-            executor.execute(new Runnable() {
+            executor.execute(new Runnable() {
-
+
-                @Override
+                @Override
-                public void run() {
+                public void run() {
-                    handler.onSubscribe(new InterceptSubscribeMessage(sub, username));
+                    handler.onSubscribe(new InterceptSubscribeMessage(sub, username));
-                }
+                }
-            });
+            });
-        }
+        }
-    }
+    }
-
+
-    @Override
+    @Override
-    public void notifyTopicUnsubscribed(final String topic, final String clientID, final String username) {
+    public void notifyTopicUnsubscribed(final String topic, final String clientID, final String username) {
-        for (final InterceptHandler handler : this.handlers.get(InterceptUnsubscribeMessage.class)) {
+        for (final InterceptHandler handler : this.handlers.get(InterceptUnsubscribeMessage.class)) {
-            LOG.debug(
+            LOG.debug(
-                    "Notifying MQTT UNSUBSCRIBE message to interceptor. MqttClientId = {}, topic = {}, "
+                    "Notifying MQTT UNSUBSCRIBE message to interceptor. MqttClientId = {}, topic = {}, "
-                            + "interceptorId = {}.",
+                            + "interceptorId = {}.",
-                    clientID,
+                    clientID,
-                    topic,
+                    topic,
-                    handler.getID());
+                    handler.getID());
-            executor.execute(new Runnable() {
+            executor.execute(new Runnable() {
-
+
-                @Override
+                @Override
-                public void run() {
+                public void run() {
-                    handler.onUnsubscribe(new InterceptUnsubscribeMessage(topic, clientID, username));
+                    handler.onUnsubscribe(new InterceptUnsubscribeMessage(topic, clientID, username));
-                }
+                }
-            });
+            });
-        }
+        }
-    }
+    }
-
+
-    @Override
+    @Override
-    public void notifyMessageAcknowledged(final InterceptAcknowledgedMessage msg) {
+    public void notifyMessageAcknowledged(final InterceptAcknowledgedMessage msg) {
-        for (final InterceptHandler handler : this.handlers.get(InterceptAcknowledgedMessage.class)) {
+        for (final InterceptHandler handler : this.handlers.get(InterceptAcknowledgedMessage.class)) {
-            LOG.debug(
+            LOG.debug(
-                    "Notifying MQTT ACK message to interceptor. MqttClientId = {}, messageId = {}, topic = {}, "
+                    "Notifying MQTT ACK message to interceptor. MqttClientId = {}, messageId = {}, topic = {}, "
-                            + "interceptorId = {}.",
+                            + "interceptorId = {}.",
-                    msg.getMsg().getClientID(),
+                    msg.getMsg().getClientID(),
-                    msg.getMsg().getMessageID(),
+                    msg.getMsg().getMessageID(),
-                    msg.getTopic(),
+                    msg.getTopic(),
-                    handler.getID());
+                    handler.getID());
-            executor.execute(new Runnable() {
+            executor.execute(new Runnable() {
-
+
-                @Override
+                @Override
-                public void run() {
+                public void run() {
-                    handler.onMessageAcknowledged(msg);
+                    handler.onMessageAcknowledged(msg);
-                }
+                }
-            });
+            });
-        }
+        }
-    }
+    }
-
+
-    @Override
+    @Override
-    public void addInterceptHandler(InterceptHandler interceptHandler) {
+    public void addInterceptHandler(InterceptHandler interceptHandler) {
-        Class<?>[] interceptedMessageTypes = getInterceptedMessageTypes(interceptHandler);
+        Class<?>[] interceptedMessageTypes = getInterceptedMessageTypes(interceptHandler);
-        LOG.info(
+        LOG.info(
-                "Adding MQTT message interceptor. InterceptorId = {}, handledMessageTypes = {}.",
+                "Adding MQTT message interceptor. InterceptorId = {}, handledMessageTypes = {}.",
-                interceptHandler.getID(),
+                interceptHandler.getID(),
-                interceptedMessageTypes);
+                interceptedMessageTypes);
-        for (Class<?> interceptMessageType : interceptedMessageTypes) {
+        for (Class<?> interceptMessageType : interceptedMessageTypes) {
-            this.handlers.get(interceptMessageType).add(interceptHandler);
+            this.handlers.get(interceptMessageType).add(interceptHandler);
-        }
+        }
-    }
+    }
-
+
-    @Override
+    @Override
-    public void removeInterceptHandler(InterceptHandler interceptHandler) {
+    public void removeInterceptHandler(InterceptHandler interceptHandler) {
-        Class<?>[] interceptedMessageTypes = getInterceptedMessageTypes(interceptHandler);
+        Class<?>[] interceptedMessageTypes = getInterceptedMessageTypes(interceptHandler);
-        LOG.info(
+        LOG.info(
-                "Removing MQTT message interceptor. InterceptorId = {}, handledMessageTypes = {}.",
+                "Removing MQTT message interceptor. InterceptorId = {}, handledMessageTypes = {}.",
-                interceptHandler.getID(),
+                interceptHandler.getID(),
-                interceptedMessageTypes);
+                interceptedMessageTypes);
-        for (Class<?> interceptMessageType : interceptedMessageTypes) {
+        for (Class<?> interceptMessageType : interceptedMessageTypes) {
-            this.handlers.get(interceptMessageType).remove(interceptHandler);
+            this.handlers.get(interceptMessageType).remove(interceptHandler);
-        }
+        }
-    }
+    }
-
+
-    private static Class<?>[] getInterceptedMessageTypes(InterceptHandler interceptHandler) {
+    private static Class<?>[] getInterceptedMessageTypes(InterceptHandler interceptHandler) {
-        Class<?>[] interceptedMessageTypes = interceptHandler.getInterceptedMessageTypes();
+        Class<?>[] interceptedMessageTypes = interceptHandler.getInterceptedMessageTypes();
-        if (interceptedMessageTypes == null) {
+        if (interceptedMessageTypes == null) {
-            return InterceptHandler.ALL_MESSAGE_TYPES;
+            return InterceptHandler.ALL_MESSAGE_TYPES;
-        }
+        }
-        return interceptedMessageTypes;
+        return interceptedMessageTypes;
-    }
+    }
-}
+}
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -1,348 +1,348 @@
-/*
+/*
- * 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
+ * All rights reserved. This program and the accompanying materials
- * are made available under the terms of the Eclipse Public License v1.0
+ * are made available under the terms of the Eclipse Public License v1.0
- * and Apache License v2.0 which accompanies this distribution.
+ * and Apache License v2.0 which accompanies this distribution.
- *
+ *
- * The Eclipse Public License is available at
+ * The Eclipse Public License is available at
- * http://www.eclipse.org/legal/epl-v10.html
+ * http://www.eclipse.org/legal/epl-v10.html
- *
+ *
- * The Apache License v2.0 is available at
+ * The Apache License v2.0 is available at
- * http://www.opensource.org/licenses/apache2.0.php
+ * http://www.opensource.org/licenses/apache2.0.php
- *
+ *
- * You may elect to redistribute this code under either of these licenses.
+ * You may elect to redistribute this code under either of these licenses.
- */
+ */
-
+
-package io.moquette.spi.persistence;
+package io.moquette.spi.persistence;
-
+
-import io.moquette.spi.ClientSession;
+import io.moquette.spi.ClientSession;
-import io.moquette.spi.IMessagesStore.StoredMessage;
+import io.moquette.spi.IMessagesStore.StoredMessage;
-import io.moquette.spi.ISessionsStore;
+import io.moquette.spi.ISessionsStore;
-import io.moquette.spi.MessageGUID;
+import io.moquette.spi.MessageGUID;
-import io.moquette.spi.impl.Utils;
+import io.moquette.spi.impl.Utils;
-import io.moquette.spi.impl.subscriptions.Subscription;
+import io.moquette.spi.impl.subscriptions.Subscription;
-import io.moquette.spi.impl.subscriptions.Topic;
+import io.moquette.spi.impl.subscriptions.Topic;
-import io.moquette.spi.persistence.MapDBPersistentStore.PersistentSession;
+import io.moquette.spi.persistence.MapDBPersistentStore.PersistentSession;
-import org.mapdb.DB;
+import org.mapdb.DB;
-import org.slf4j.Logger;
+import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.slf4j.LoggerFactory;
-import java.util.*;
+import java.util.*;
-import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ConcurrentMap;
-
+
-/**
+/**
- * ISessionsStore implementation backed by MapDB.
+ * ISessionsStore implementation backed by MapDB.
- *
+ *
- * @author andrea
+ * @author andrea
- */
+ */
-class MapDBSessionsStore implements ISessionsStore {
+class MapDBSessionsStore implements ISessionsStore {
-
+
-    private static final Logger LOG = LoggerFactory.getLogger(MapDBSessionsStore.class);
+    private static final Logger LOG = LoggerFactory.getLogger(MapDBSessionsStore.class);
-
+
-    // maps clientID->[MessageId -> guid]
+    // maps clientID->[MessageId -> guid]
-    private ConcurrentMap<String, ConcurrentMap<Integer, MessageGUID>> m_inflightStore;
+    private ConcurrentMap<String, ConcurrentMap<Integer, MessageGUID>> m_inflightStore;
-    // map clientID <-> set of currently in flight packet identifiers
+    // map clientID <-> set of currently in flight packet identifiers
-    private Map<String, Set<Integer>> m_inFlightIds;
+    private Map<String, Set<Integer>> m_inFlightIds;
-    private ConcurrentMap<String, PersistentSession> m_persistentSessions;
+    private ConcurrentMap<String, PersistentSession> m_persistentSessions;
-    // maps clientID->[MessageId -> guid]
+    // maps clientID->[MessageId -> guid]
-    private ConcurrentMap<String, Map<Integer, MessageGUID>> m_secondPhaseStore;
+    private ConcurrentMap<String, Map<Integer, MessageGUID>> m_secondPhaseStore;
-
+
-    private final DB m_db;
+    private final DB m_db;
-    private final MapDBMessagesStore m_messagesStore;
+    private final MapDBMessagesStore m_messagesStore;
-
+
-    MapDBSessionsStore(DB db, MapDBMessagesStore messagesStore) {
+    MapDBSessionsStore(DB db, MapDBMessagesStore messagesStore) {
-        m_db = db;
+        m_db = db;
-        m_messagesStore = messagesStore;
+        m_messagesStore = messagesStore;
-    }
+    }
-
+
-    @Override
+    @Override
-    public void initStore() {
+    public void initStore() {
-        m_inflightStore = m_db.getHashMap("inflight");
+        m_inflightStore = m_db.getHashMap("inflight");
-        m_inFlightIds = m_db.getHashMap("inflightPacketIDs");
+        m_inFlightIds = m_db.getHashMap("inflightPacketIDs");
-        m_persistentSessions = m_db.getHashMap("sessions");
+        m_persistentSessions = m_db.getHashMap("sessions");
-        m_secondPhaseStore = m_db.getHashMap("secondPhase");
+        m_secondPhaseStore = m_db.getHashMap("secondPhase");
-    }
+    }
-
+
-    @Override
+    @Override
-    public void addNewSubscription(Subscription newSubscription) {
+    public void addNewSubscription(Subscription newSubscription) {
-        LOG.info(
+        LOG.info(
-                "Adding new subscription. CId= {}, topics = {}.",
+                "Adding new subscription. CId= {}, topics = {}.",
-                newSubscription.getClientId(),
+                newSubscription.getClientId(),
-                newSubscription.getTopicFilter());
+                newSubscription.getTopicFilter());
-        final String clientID = newSubscription.getClientId();
+        final String clientID = newSubscription.getClientId();
-        m_db.getHashMap("subscriptions_" + clientID).put(newSubscription.getTopicFilter(), newSubscription);
+        m_db.getHashMap("subscriptions_" + clientID).put(newSubscription.getTopicFilter(), newSubscription);
-
+
-        if (LOG.isTraceEnabled()) {
+        if (LOG.isTraceEnabled()) {
-            LOG.trace(
+            LOG.trace(
-                    "The subscription has been added. CId= {}, topics = {}, clientSubscriptions = {}.",
+                    "The subscription has been added. CId= {}, topics = {}, clientSubscriptions = {}.",
-                    newSubscription.getClientId(),
+                    newSubscription.getClientId(),
-                    newSubscription.getTopicFilter(),
+                    newSubscription.getTopicFilter(),
-                    m_db.getHashMap("subscriptions_" + clientID));
+                    m_db.getHashMap("subscriptions_" + clientID));
-        }
+        }
-    }
+    }
-
+
-    @Override
+    @Override
-    public void removeSubscription(Topic topicFilter, String clientID) {
+    public void removeSubscription(Topic topicFilter, String clientID) {
-        LOG.info("Removing subscription. CId= {}, topics = {}.", clientID, topicFilter);
+        LOG.info("Removing subscription. CId= {}, topics = {}.", clientID, topicFilter);
-        if (!m_db.exists("subscriptions_" + clientID)) {
+        if (!m_db.exists("subscriptions_" + clientID)) {
-            return;
+            return;
-        }
+        }
-        m_db.getHashMap("subscriptions_" + clientID).remove(topicFilter);
+        m_db.getHashMap("subscriptions_" + clientID).remove(topicFilter);
-        if (LOG.isDebugEnabled()) {
+        if (LOG.isDebugEnabled()) {
-            LOG.debug(
+            LOG.debug(
-                    "The subscription has been removed. CId= {}, topics = {}, clientSubscriptions = {}.",
+                    "The subscription has been removed. CId= {}, topics = {}, clientSubscriptions = {}.",
-                    clientID,
+                    clientID,
-                    topicFilter,
+                    topicFilter,
-                    m_db.getHashMap("subscriptions_" + clientID));
+                    m_db.getHashMap("subscriptions_" + clientID));
-        }
+        }
-    }
+    }
-
+
-    @Override
+    @Override
-    public void wipeSubscriptions(String clientID) {
+    public void wipeSubscriptions(String clientID) {
-        LOG.info("Wiping subscriptions. CId= {}.", clientID);
+        LOG.info("Wiping subscriptions. CId= {}.", clientID);
-        m_db.delete("subscriptions_" + clientID);
+        m_db.delete("subscriptions_" + clientID);
-        if (LOG.isDebugEnabled()) {
+        if (LOG.isDebugEnabled()) {
-            LOG.debug(
+            LOG.debug(
-                    "The subscriptions have been removed. CId= {}, clientSubscriptions = {}.",
+                    "The subscriptions have been removed. CId= {}, clientSubscriptions = {}.",
-                    clientID,
+                    clientID,
-                    m_db.getHashMap("subscriptions_" + clientID));
+                    m_db.getHashMap("subscriptions_" + clientID));
-        }
+        }
-    }
+    }
-
+
-    @Override
+    @Override
-    public List<ClientTopicCouple> listAllSubscriptions() {
+    public List<ClientTopicCouple> listAllSubscriptions() {
-        LOG.info("Retrieving existing subscriptions...");
+        LOG.info("Retrieving existing subscriptions...");
-        final List<ClientTopicCouple> allSubscriptions = new ArrayList<>();
+        final List<ClientTopicCouple> allSubscriptions = new ArrayList<>();
-        for (String clientID : m_persistentSessions.keySet()) {
+        for (String clientID : m_persistentSessions.keySet()) {
-            ConcurrentMap<Topic, Subscription> clientSubscriptions = m_db.getHashMap("subscriptions_" + clientID);
+            ConcurrentMap<Topic, Subscription> clientSubscriptions = m_db.getHashMap("subscriptions_" + clientID);
-            for (Topic topicFilter : clientSubscriptions.keySet()) {
+            for (Topic topicFilter : clientSubscriptions.keySet()) {
-                allSubscriptions.add(new ClientTopicCouple(clientID, topicFilter));
+                allSubscriptions.add(new ClientTopicCouple(clientID, topicFilter));
-            }
+            }
-        }
+        }
-        if (LOG.isTraceEnabled()) {
+        if (LOG.isTraceEnabled()) {
-            LOG.trace("The existing subscriptions have been retrieved. Result = {}.", allSubscriptions);
+            LOG.trace("The existing subscriptions have been retrieved. Result = {}.", allSubscriptions);
-        }
+        }
-        return allSubscriptions;
+        return allSubscriptions;
-    }
+    }
-
+
-    @Override
+    @Override
-    public Subscription getSubscription(ClientTopicCouple couple) {
+    public Subscription getSubscription(ClientTopicCouple couple) {
-        ConcurrentMap<Topic, Subscription> clientSubscriptions = m_db.getHashMap("subscriptions_" + couple.clientID);
+        ConcurrentMap<Topic, Subscription> clientSubscriptions = m_db.getHashMap("subscriptions_" + couple.clientID);
-        LOG.info("Retrieving subscriptions. CId= {}, subscriptions = {}.", couple.clientID, clientSubscriptions);
+        LOG.info("Retrieving subscriptions. CId= {}, subscriptions = {}.", couple.clientID, clientSubscriptions);
-        return clientSubscriptions.get(couple.topicFilter);
+        return clientSubscriptions.get(couple.topicFilter);
-    }
+    }
-
+
-    @Override
+    @Override
-    public List<Subscription> getSubscriptions() {
+    public List<Subscription> getSubscriptions() {
-        LOG.info("Retrieving existing subscriptions...");
+        LOG.info("Retrieving existing subscriptions...");
-        List<Subscription> subscriptions = new ArrayList<>();
+        List<Subscription> subscriptions = new ArrayList<>();
-        for (String clientID : m_persistentSessions.keySet()) {
+        for (String clientID : m_persistentSessions.keySet()) {
-            ConcurrentMap<Topic, Subscription> clientSubscriptions = m_db.getHashMap("subscriptions_" + clientID);
+            ConcurrentMap<Topic, Subscription> clientSubscriptions = m_db.getHashMap("subscriptions_" + clientID);
-            subscriptions.addAll(clientSubscriptions.values());
+            subscriptions.addAll(clientSubscriptions.values());
-        }
+        }
-        LOG.debug("The existing subscriptions have been retrieved. Result = {}.", subscriptions);
+        LOG.debug("The existing subscriptions have been retrieved. Result = {}.", subscriptions);
-        return subscriptions;
+        return subscriptions;
-    }
+    }
-
+
-    @Override
+    @Override
-    public boolean contains(String clientID) {
+    public boolean contains(String clientID) {
-        return m_db.exists("subscriptions_" + clientID);
+        return m_db.exists("subscriptions_" + clientID);
-    }
+    }
-
+
-    @Override
+    @Override
-    public ClientSession createNewSession(String clientID, boolean cleanSession) {
+    public ClientSession createNewSession(String clientID, boolean cleanSession) {
-        if (m_persistentSessions.containsKey(clientID)) {
+        if (m_persistentSessions.containsKey(clientID)) {
-            LOG.error(
+            LOG.error(
-                    "Unable to create a new session: the client ID is already in use. CId= {}, cleanSession = {}.",
+                    "Unable to create a new session: the client ID is already in use. CId= {}, cleanSession = {}.",
-                    clientID,
+                    clientID,
-                    cleanSession);
+                    cleanSession);
-            throw new IllegalArgumentException("Can't create a session with the ID of an already existing" + clientID);
+            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);
+        LOG.info("Creating new session. CId= {}, cleanSession = {}.", clientID, cleanSession);
-        m_persistentSessions.putIfAbsent(clientID, new PersistentSession(cleanSession));
+        m_persistentSessions.putIfAbsent(clientID, new PersistentSession(cleanSession));
-        return new ClientSession(clientID, m_messagesStore, this, cleanSession);
+        return new ClientSession(clientID, m_messagesStore, this, cleanSession);
-    }
+    }
-
+
-    @Override
+    @Override
-    public ClientSession sessionForClient(String clientID) {
+    public ClientSession sessionForClient(String clientID) {
-        LOG.info("Retrieving session. CId= {}.", clientID);
+        LOG.info("Retrieving session. CId= {}.", clientID);
-        if (!m_persistentSessions.containsKey(clientID)) {
+        if (!m_persistentSessions.containsKey(clientID)) {
-            LOG.warn("The session does not exist. CId= {}.", clientID);
+            LOG.warn("The session does not exist. CId= {}.", clientID);
-            return null;
+            return null;
-        }
+        }
-
+
-        PersistentSession storedSession = m_persistentSessions.get(clientID);
+        PersistentSession storedSession = m_persistentSessions.get(clientID);
-        return new ClientSession(clientID, m_messagesStore, this, storedSession.cleanSession);
+        return new ClientSession(clientID, m_messagesStore, this, storedSession.cleanSession);
-    }
+    }
-
+
-    @Override
+    @Override
-    public Collection<ClientSession> getAllSessions() {
+    public Collection<ClientSession> getAllSessions() {
-        Collection<ClientSession> result = new ArrayList<>();
+        Collection<ClientSession> result = new ArrayList<>();
-        for (Map.Entry<String, PersistentSession> entry : m_persistentSessions.entrySet()) {
+        for (Map.Entry<String, PersistentSession> entry : m_persistentSessions.entrySet()) {
-            result.add(new ClientSession(entry.getKey(), m_messagesStore, this, entry.getValue().cleanSession));
+            result.add(new ClientSession(entry.getKey(), m_messagesStore, this, entry.getValue().cleanSession));
-        }
+        }
-        return result;
+        return result;
-    }
+    }
-
+
-    @Override
+    @Override
-    public void updateCleanStatus(String clientID, boolean cleanSession) {
+    public void updateCleanStatus(String clientID, boolean cleanSession) {
-        LOG.info("Updating cleanSession flag. CId= {}, cleanSession = {}.", clientID, cleanSession);
+        LOG.info("Updating cleanSession flag. CId= {}, cleanSession = {}.", clientID, cleanSession);
-        m_persistentSessions.put(clientID, new MapDBPersistentStore.PersistentSession(cleanSession));
+        m_persistentSessions.put(clientID, new MapDBPersistentStore.PersistentSession(cleanSession));
-    }
+    }
-
+
-    /**
+    /**
-     * Return the next valid packetIdentifier for the given client session.
+     * Return the next valid packetIdentifier for the given client session.
-     */
+     */
-    @Override
+    @Override
-    public int nextPacketID(String clientID) {
+    public int nextPacketID(String clientID) {
-        if (LOG.isDebugEnabled()) {
+        if (LOG.isDebugEnabled()) {
-            LOG.debug("Generating next packet ID. CId= {}.", clientID);
+            LOG.debug("Generating next packet ID. CId= {}.", clientID);
-        }
+        }
-        Set<Integer> inFlightForClient = this.m_inFlightIds.get(clientID);
+        Set<Integer> inFlightForClient = this.m_inFlightIds.get(clientID);
-        if (inFlightForClient == null) {
+        if (inFlightForClient == null) {
-            int nextPacketId = 1;
+            int nextPacketId = 1;
-            inFlightForClient = Collections.newSetFromMap(new ConcurrentHashMap<Integer, Boolean>());
+            inFlightForClient = Collections.newSetFromMap(new ConcurrentHashMap<Integer, Boolean>());
-            inFlightForClient.add(nextPacketId);
+            inFlightForClient.add(nextPacketId);
-            this.m_inFlightIds.put(clientID, inFlightForClient);
+            this.m_inFlightIds.put(clientID, inFlightForClient);
-            return nextPacketId;
+            return nextPacketId;
-
+
-        }
+        }
-
+
-        int maxId = inFlightForClient.isEmpty() ? 0 : Collections.max(inFlightForClient);
+        int maxId = inFlightForClient.isEmpty() ? 0 : Collections.max(inFlightForClient);
-        int nextPacketId = (maxId + 1) % 0xFFFF;
+        int nextPacketId = (maxId + 1) % 0xFFFF;
-        inFlightForClient.add(nextPacketId);
+        inFlightForClient.add(nextPacketId);
-        if (LOG.isDebugEnabled()) {
+        if (LOG.isDebugEnabled()) {
-            LOG.debug("The next packet ID has been generated. CId= {}, result = {}.", clientID, nextPacketId);
+            LOG.debug("The next packet ID has been generated. CId= {}, result = {}.", clientID, nextPacketId);
-        }
+        }
-        return nextPacketId;
+        return nextPacketId;
-    }
+    }
-
+
-    @Override
+    @Override
-    public void inFlightAck(String clientID, int messageID) {
+    public void inFlightAck(String clientID, int messageID) {
-        LOG.debug("Acknowledging inflight message. CId= {}, messageId = {}.", clientID, messageID);
+        LOG.debug("Acknowledging inflight message. CId= {}, messageId = {}.", clientID, messageID);
-        Map<Integer, MessageGUID> m = this.m_inflightStore.get(clientID);
+        Map<Integer, MessageGUID> m = this.m_inflightStore.get(clientID);
-        if (m == null) {
+        if (m == null) {
-            LOG.warn("Unable to retrieve inflight message record. CId= {}, messageId = {}.", clientID, messageID);
+            LOG.warn("Unable to retrieve inflight message record. CId= {}, messageId = {}.", clientID, messageID);
-            return;
+            return;
-        }
+        }
-        m.remove(messageID);
+        m.remove(messageID);
-
+
-        // remove from the ids store
+        // remove from the ids store
-        Set<Integer> inFlightForClient = this.m_inFlightIds.get(clientID);
+        Set<Integer> inFlightForClient = this.m_inFlightIds.get(clientID);
-        if (inFlightForClient != null) {
+        if (inFlightForClient != null) {
-            inFlightForClient.remove(messageID);
+            inFlightForClient.remove(messageID);
-        }
+        }
-    }
+    }
-
+
-    @Override
+    @Override
-    public void inFlight(String clientID, int messageID, MessageGUID guid) {
+    public void inFlight(String clientID, int messageID, MessageGUID guid) {
-        LOG.debug("Storing inflight message. CId= {}, messageId = {}, guid = {}.", clientID, messageID, guid);
+        LOG.debug("Storing inflight message. CId= {}, messageId = {}, guid = {}.", clientID, messageID, guid);
-        ConcurrentMap<Integer, MessageGUID> m = this.m_inflightStore.get(clientID);
+        ConcurrentMap<Integer, MessageGUID> m = this.m_inflightStore.get(clientID);
-        if (m == null) {
+        if (m == null) {
-            m = new ConcurrentHashMap<>();
+            m = new ConcurrentHashMap<>();
-        }
+        }
-        m.put(messageID, guid);
+        m.put(messageID, guid);
-        LOG.info("storing inflight clientID <{}> messageID {} guid <{}>", clientID, messageID, guid);
+        LOG.info("storing inflight clientID <{}> messageID {} guid <{}>", clientID, messageID, guid);
-        this.m_inflightStore.put(clientID, m);
+        this.m_inflightStore.put(clientID, m);
-    }
+    }
-
+
-    @Override
+    @Override
-    public BlockingQueue<StoredMessage> queue(String clientID) {
+    public BlockingQueue<StoredMessage> queue(String clientID) {
-        LOG.info("Queuing pending message. CId= {}, guid = {}.", clientID);
+        LOG.info("Queuing pending message. CId= {}, guid = {}.", clientID);
-        return this.m_db.getQueue(clientID);
+        return this.m_db.getQueue(clientID);
-    }
+    }
-
+
-    @Override
+    @Override
-    public void dropQueue(String clientID) {
+    public void dropQueue(String clientID) {
-        LOG.info("Removing pending messages. CId= {}.", clientID);
+        LOG.info("Removing pending messages. CId= {}.", clientID);
-        this.m_db.delete(clientID);
+        this.m_db.delete(clientID);
-    }
+    }
-
+
-    @Override
+    @Override
-    public void moveInFlightToSecondPhaseAckWaiting(String clientID, int messageID) {
+    public void moveInFlightToSecondPhaseAckWaiting(String clientID, int messageID) {
-        if (LOG.isDebugEnabled()) {
+        if (LOG.isDebugEnabled()) {
-            LOG.debug("Moving inflight message to 2nd phase ack state. CId= {}, messageID = {}.", clientID, messageID);
+            LOG.debug("Moving inflight message to 2nd phase ack state. CId= {}, messageID = {}.", clientID, messageID);
-        }
+        }
-        Map<Integer, MessageGUID> m = this.m_inflightStore.get(clientID);
+        Map<Integer, MessageGUID> m = this.m_inflightStore.get(clientID);
-        if (m == null) {
+        if (m == null) {
-            LOG.warn("Unable to retrieve inflight message record. CId= {}, messageId = {}.", clientID, messageID);
+            LOG.warn("Unable to retrieve inflight message record. CId= {}, messageId = {}.", clientID, messageID);
-            return;
+            return;
-        }
+        }
-        MessageGUID guid = m.remove(messageID);
+        MessageGUID guid = m.remove(messageID);
-
+
-        // remove from the ids store
+        // remove from the ids store
-        Set<Integer> inFlightForClient = this.m_inFlightIds.get(clientID);
+        Set<Integer> inFlightForClient = this.m_inFlightIds.get(clientID);
-        if (inFlightForClient != null) {
+        if (inFlightForClient != null) {
-            inFlightForClient.remove(messageID);
+            inFlightForClient.remove(messageID);
-        }
+        }
-
+
-        Map<Integer, MessageGUID> messageIDs = Utils
+        Map<Integer, MessageGUID> messageIDs = Utils
-                .defaultGet(m_secondPhaseStore, clientID, new HashMap<Integer, MessageGUID>());
+                .defaultGet(m_secondPhaseStore, clientID, new HashMap<Integer, MessageGUID>());
-        messageIDs.put(messageID, guid);
+        messageIDs.put(messageID, guid);
-        m_secondPhaseStore.put(clientID, messageIDs);
+        m_secondPhaseStore.put(clientID, messageIDs);
-    }
+    }
-
+
-    @Override
+    @Override
-    public MessageGUID secondPhaseAcknowledged(String clientID, int messageID) {
+    public MessageGUID secondPhaseAcknowledged(String clientID, int messageID) {
-        if (LOG.isDebugEnabled()) {
+        if (LOG.isDebugEnabled()) {
-            LOG.debug("Processing second phase ACK CId={}, messageId={}", clientID, messageID);
+            LOG.debug("Processing second phase ACK CId={}, messageId={}", clientID, messageID);
-        }
+        }
-        Map<Integer, MessageGUID> messageIDs = Utils
+        Map<Integer, MessageGUID> messageIDs = Utils
-                .defaultGet(m_secondPhaseStore, clientID, new HashMap<Integer, MessageGUID>());
+                .defaultGet(m_secondPhaseStore, clientID, new HashMap<Integer, MessageGUID>());
-        MessageGUID guid = messageIDs.remove(messageID);
+        MessageGUID guid = messageIDs.remove(messageID);
-        m_secondPhaseStore.put(clientID, messageIDs);
+        m_secondPhaseStore.put(clientID, messageIDs);
-        return guid;
+        return guid;
-    }
+    }
-
+
-    @Override
+    @Override
-    public MessageGUID mapToGuid(String clientID, int messageID) {
+    public MessageGUID mapToGuid(String clientID, int messageID) {
-        LOG.debug("Mapping message ID to GUID CId={}, messageId={}", clientID, messageID);
+        LOG.debug("Mapping message ID to GUID CId={}, messageId={}", clientID, messageID);
-        ConcurrentMap<Integer, MessageGUID> messageIdToGuid = m_db.getHashMap(messageId2GuidsMapName(clientID));
+        ConcurrentMap<Integer, MessageGUID> messageIdToGuid = m_db.getHashMap(messageId2GuidsMapName(clientID));
-        MessageGUID result = messageIdToGuid.get(messageID);
+        MessageGUID result = messageIdToGuid.get(messageID);
-        LOG.debug("Message ID has been mapped to a GUID CId={}, messageId={}, guid={}", clientID, messageID, result);
+        LOG.debug("Message ID has been mapped to a GUID CId={}, messageId={}, guid={}", clientID, messageID, result);
-        return result;
+        return result;
-    }
+    }
-
+
-    static String messageId2GuidsMapName(String clientID) {
+    static String messageId2GuidsMapName(String clientID) {
-        return "guidsMapping_" + clientID;
+        return "guidsMapping_" + clientID;
-    }
+    }
-
+
-    @Override
+    @Override
-    public StoredMessage getInflightMessage(String clientID, int messageID) {
+    public StoredMessage getInflightMessage(String clientID, int messageID) {
-        LOG.info("Retrieving inflight message CId={}, messageId={}", clientID, messageID);
+        LOG.info("Retrieving inflight message CId={}, messageId={}", clientID, messageID);
-        Map<Integer, MessageGUID> clientEntries = m_inflightStore.get(clientID);
+        Map<Integer, MessageGUID> clientEntries = m_inflightStore.get(clientID);
-        if (clientEntries == null) {
+        if (clientEntries == null) {
-            LOG.warn("The client has no inflight messages CId={}, messageId={}", clientID, messageID);
+            LOG.warn("The client has no inflight messages CId={}, messageId={}", clientID, messageID);
-            return null;
+            return null;
-        }
+        }
-        MessageGUID guid = clientEntries.get(messageID);
+        MessageGUID guid = clientEntries.get(messageID);
-        if (guid == null) {
+        if (guid == null) {
-            LOG.warn("The message ID does not have an associated GUID. CId= {}, messageId = {}.", clientID, messageID);
+            LOG.warn("The message ID does not have an associated GUID. CId= {}, messageId = {}.", clientID, messageID);
-            return null;
+            return null;
-        }
+        }
-        return m_messagesStore.getMessageByGuid(guid);
+        return m_messagesStore.getMessageByGuid(guid);
-    }
+    }
-
+
-    @Override
+    @Override
-    public int getInflightMessagesNo(String clientID) {
+    public int getInflightMessagesNo(String clientID) {
-        if (!m_inflightStore.containsKey(clientID))
+        if (!m_inflightStore.containsKey(clientID))
-            return 0;
+            return 0;
-        else
+        else
-            return m_inflightStore.get(clientID).size();
+            return m_inflightStore.get(clientID).size();
-    }
+    }
-
+
-    @Override
+    @Override
-    public int getPendingPublishMessagesNo(String clientID) {
+    public int getPendingPublishMessagesNo(String clientID) {
-        return m_messagesStore.getPendingPublishMessages(clientID);
+        return m_messagesStore.getPendingPublishMessages(clientID);
-    }
+    }
-
+
-    @Override
+    @Override
-    public int getSecondPhaseAckPendingMessages(String clientID) {
+    public int getSecondPhaseAckPendingMessages(String clientID) {
-        if (!m_secondPhaseStore.containsKey(clientID))
+        if (!m_secondPhaseStore.containsKey(clientID))
-            return 0;
+            return 0;
-        else
+        else
-            return m_secondPhaseStore.get(clientID).size();
+            return m_secondPhaseStore.get(clientID).size();
-    }
+    }
-}
+}
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -1,174 +1,174 @@
-/*
+/*
- * 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
+ * All rights reserved. This program and the accompanying materials
- * are made available under the terms of the Eclipse Public License v1.0
+ * are made available under the terms of the Eclipse Public License v1.0
- * and Apache License v2.0 which accompanies this distribution.
+ * and Apache License v2.0 which accompanies this distribution.
- *
+ *
- * The Eclipse Public License is available at
+ * The Eclipse Public License is available at
- * http://www.eclipse.org/legal/epl-v10.html
+ * http://www.eclipse.org/legal/epl-v10.html
- *
+ *
- * The Apache License v2.0 is available at
+ * The Apache License v2.0 is available at
- * http://www.opensource.org/licenses/apache2.0.php
+ * http://www.opensource.org/licenses/apache2.0.php
- *
+ *
- * You may elect to redistribute this code under either of these licenses.
+ * You may elect to redistribute this code under either of these licenses.
- */
+ */
-
+
-package io.moquette.spi.impl;
+package io.moquette.spi.impl;
-
+
-import io.moquette.interception.InterceptHandler;
+import io.moquette.interception.InterceptHandler;
-import io.moquette.interception.messages.*;
+import io.moquette.interception.messages.*;
-import io.moquette.server.netty.MessageBuilder;
+import io.moquette.server.netty.MessageBuilder;
-import io.moquette.spi.impl.subscriptions.Subscription;
+import io.moquette.spi.impl.subscriptions.Subscription;
-import io.moquette.spi.impl.subscriptions.Topic;
+import io.moquette.spi.impl.subscriptions.Topic;
-import io.netty.handler.codec.mqtt.MqttQoS;
+import io.netty.handler.codec.mqtt.MqttQoS;
-import org.junit.AfterClass;
+import org.junit.AfterClass;
-import org.junit.BeforeClass;
+import org.junit.BeforeClass;
-import org.junit.Test;
+import org.junit.Test;
-import java.util.Arrays;
+import java.util.Arrays;
-import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicInteger;
-import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertEquals;
-import static org.mockito.Matchers.refEq;
+import static org.mockito.Matchers.refEq;
-import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verify;
-
+
-/**
+/**
- * @author Wagner Macedo
+ * @author Wagner Macedo
- */
+ */
-public class BrokerInterceptorTest {
+public class BrokerInterceptorTest {
-
+
-    // value to check for changes after every notification
+    // value to check for changes after every notification
-    private static final AtomicInteger n = new AtomicInteger(0);
+    private static final AtomicInteger n = new AtomicInteger(0);
-
+
-    // Interceptor loaded with a custom InterceptHandler special for the tests
+    // Interceptor loaded with a custom InterceptHandler special for the tests
-    private static final class MockObserver implements InterceptHandler {
+    private static final class MockObserver implements InterceptHandler {
-
+
-        @Override
+        @Override
-        public String getID() {
+        public String getID() {
-            return "MockObserver";
+            return "MockObserver";
-        }
+        }
-
+
-        @Override
+        @Override
-        public Class<?>[] getInterceptedMessageTypes() {
+        public Class<?>[] getInterceptedMessageTypes() {
-            return InterceptHandler.ALL_MESSAGE_TYPES;
+            return InterceptHandler.ALL_MESSAGE_TYPES;
-        }
+        }
-
+
-        @Override
+        @Override
-        public void onConnect(InterceptConnectMessage msg) {
+        public void onConnect(InterceptConnectMessage msg) {
-            n.set(40);
+            n.set(40);
-        }
+        }
-
+
-        @Override
+        @Override
-        public void onDisconnect(InterceptDisconnectMessage msg) {
+        public void onDisconnect(InterceptDisconnectMessage msg) {
-            n.set(50);
+            n.set(50);
-        }
+        }
-
+
-        @Override
+        @Override
-        public void onConnectionLost(InterceptConnectionLostMessage msg) {
+        public void onConnectionLost(InterceptConnectionLostMessage msg) {
-            n.set(30);
+            n.set(30);
-        }
+        }
-
+
-        @Override
+        @Override
-        public void onPublish(InterceptPublishMessage msg) {
+        public void onPublish(InterceptPublishMessage msg) {
-            n.set(60);
+            n.set(60);
-        }
+        }
-
+
-        @Override
+        @Override
-        public void onSubscribe(InterceptSubscribeMessage msg) {
+        public void onSubscribe(InterceptSubscribeMessage msg) {
-            n.set(70);
+            n.set(70);
-        }
+        }
-
+
-        @Override
+        @Override
-        public void onUnsubscribe(InterceptUnsubscribeMessage msg) {
+        public void onUnsubscribe(InterceptUnsubscribeMessage msg) {
-            n.set(80);
+            n.set(80);
-        }
+        }
-
+
-        @Override
+        @Override
-        public void onMessageAcknowledged(InterceptAcknowledgedMessage msg) {
+        public void onMessageAcknowledged(InterceptAcknowledgedMessage msg) {
-            n.set(90);
+            n.set(90);
-        }
+        }
-    }
+    }
-
+
-    private static final BrokerInterceptor interceptor = new BrokerInterceptor(
+    private static final BrokerInterceptor interceptor = new BrokerInterceptor(
-            Arrays.<InterceptHandler>asList(new MockObserver()));
+            Arrays.<InterceptHandler>asList(new MockObserver()));
-
+
-    @BeforeClass
+    @BeforeClass
-    public static void beforeAllTests() {
+    public static void beforeAllTests() {
-        // check if any of the handler methods was called before notifications
+        // check if any of the handler methods was called before notifications
-        assertEquals(0, n.get());
+        assertEquals(0, n.get());
-    }
+    }
-
+
-    @AfterClass
+    @AfterClass
-    public static void afterAllTests() {
+    public static void afterAllTests() {
-        interceptor.stop();
+        interceptor.stop();
-    }
+    }
-
+
-    /* Used to wait handler notification by the interceptor internal thread */
+    /* Used to wait handler notification by the interceptor internal thread */
-    private static void interval() throws InterruptedException {
+    private static void interval() throws InterruptedException {
-        Thread.sleep(100);
+        Thread.sleep(100);
-    }
+    }
-
+
-    @Test
+    @Test
-    public void testNotifyClientConnected() throws Exception {
+    public void testNotifyClientConnected() throws Exception {
-        interceptor.notifyClientConnected(MessageBuilder.connect().build());
+        interceptor.notifyClientConnected(MessageBuilder.connect().build());
-        interval();
+        interval();
-        assertEquals(40, n.get());
+        assertEquals(40, n.get());
-    }
+    }
-
+
-    @Test
+    @Test
-    public void testNotifyClientDisconnected() throws Exception {
+    public void testNotifyClientDisconnected() throws Exception {
-        interceptor.notifyClientDisconnected("cli1234", "cli1234");
+        interceptor.notifyClientDisconnected("cli1234", "cli1234");
-        interval();
+        interval();
-        assertEquals(50, n.get());
+        assertEquals(50, n.get());
-    }
+    }
-
+
-    @Test
+    @Test
-    public void testNotifyTopicPublished() throws Exception {
+    public void testNotifyTopicPublished() throws Exception {
-        interceptor.notifyTopicPublished(
+        interceptor.notifyTopicPublished(
-                MessageBuilder.publish().qos(MqttQoS.AT_MOST_ONCE).payload("Hello".getBytes()).build(),
+                MessageBuilder.publish().qos(MqttQoS.AT_MOST_ONCE).payload("Hello".getBytes()).build(),
-                "cli1234",
+                "cli1234",
-                "cli1234");
+                "cli1234");
-        interval();
+        interval();
-        assertEquals(60, n.get());
+        assertEquals(60, n.get());
-    }
+    }
-
+
-    @Test
+    @Test
-    public void testNotifyTopicSubscribed() throws Exception {
+    public void testNotifyTopicSubscribed() throws Exception {
-        interceptor.notifyTopicSubscribed(new Subscription("cli1", new Topic("o2"), MqttQoS.AT_MOST_ONCE), "cli1234");
+        interceptor.notifyTopicSubscribed(new Subscription("cli1", new Topic("o2"), MqttQoS.AT_MOST_ONCE), "cli1234");
-        interval();
+        interval();
-        assertEquals(70, n.get());
+        assertEquals(70, n.get());
-    }
+    }
-
+
-    @Test
+    @Test
-    public void testNotifyTopicUnsubscribed() throws Exception {
+    public void testNotifyTopicUnsubscribed() throws Exception {
-        interceptor.notifyTopicUnsubscribed("o2", "cli1234", "cli1234");
+        interceptor.notifyTopicUnsubscribed("o2", "cli1234", "cli1234");
-        interval();
+        interval();
-        assertEquals(80, n.get());
+        assertEquals(80, n.get());
-    }
+    }
-
+
-    @Test
+    @Test
-    public void testAddAndRemoveInterceptHandler() throws Exception {
+    public void testAddAndRemoveInterceptHandler() throws Exception {
-        InterceptHandler interceptHandlerMock1 = mock(InterceptHandler.class);
+        InterceptHandler interceptHandlerMock1 = mock(InterceptHandler.class);
-        InterceptHandler interceptHandlerMock2 = mock(InterceptHandler.class);
+        InterceptHandler interceptHandlerMock2 = mock(InterceptHandler.class);
-        // add
+        // add
-        interceptor.addInterceptHandler(interceptHandlerMock1);
+        interceptor.addInterceptHandler(interceptHandlerMock1);
-        interceptor.addInterceptHandler(interceptHandlerMock2);
+        interceptor.addInterceptHandler(interceptHandlerMock2);
-
+
-        Subscription subscription = new Subscription("cli1", new Topic("o2"), MqttQoS.AT_MOST_ONCE);
+        Subscription subscription = new Subscription("cli1", new Topic("o2"), MqttQoS.AT_MOST_ONCE);
-        interceptor.notifyTopicSubscribed(subscription, "cli1234");
+        interceptor.notifyTopicSubscribed(subscription, "cli1234");
-        interval();
+        interval();
-
+
-        verify(interceptHandlerMock1).onSubscribe(refEq(new InterceptSubscribeMessage(subscription, "cli1234")));
+        verify(interceptHandlerMock1).onSubscribe(refEq(new InterceptSubscribeMessage(subscription, "cli1234")));
-        verify(interceptHandlerMock2).onSubscribe(refEq(new InterceptSubscribeMessage(subscription, "cli1234")));
+        verify(interceptHandlerMock2).onSubscribe(refEq(new InterceptSubscribeMessage(subscription, "cli1234")));
-
+
-        // remove
+        // remove
-        interceptor.removeInterceptHandler(interceptHandlerMock1);
+        interceptor.removeInterceptHandler(interceptHandlerMock1);
-
+
-        interceptor.notifyTopicSubscribed(subscription, "cli1235");
+        interceptor.notifyTopicSubscribed(subscription, "cli1235");
-        interval();
+        interval();
-        // removeInterceptHandler() performs another interaction
+        // removeInterceptHandler() performs another interaction
-        // TODO: fix this
+        // TODO: fix this
-        // verifyNoMoreInteractions(interceptHandlerMock1);
+        // verifyNoMoreInteractions(interceptHandlerMock1);
-        verify(interceptHandlerMock2).onSubscribe(refEq(new InterceptSubscribeMessage(subscription, "cli1235")));
+        verify(interceptHandlerMock2).onSubscribe(refEq(new InterceptSubscribeMessage(subscription, "cli1235")));
-    }
+    }
-}
+}
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -1,273 +1,273 @@
-/*
+/*
- * 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
+ * All rights reserved. This program and the accompanying materials
- * are made available under the terms of the Eclipse Public License v1.0
+ * are made available under the terms of the Eclipse Public License v1.0
- * and Apache License v2.0 which accompanies this distribution.
+ * and Apache License v2.0 which accompanies this distribution.
- *
+ *
- * The Eclipse Public License is available at
+ * The Eclipse Public License is available at
- * http://www.eclipse.org/legal/epl-v10.html
+ * http://www.eclipse.org/legal/epl-v10.html
- *
+ *
- * The Apache License v2.0 is available at
+ * The Apache License v2.0 is available at
- * http://www.opensource.org/licenses/apache2.0.php
+ * http://www.opensource.org/licenses/apache2.0.php
- *
+ *
- * You may elect to redistribute this code under either of these licenses.
+ * You may elect to redistribute this code under either of these licenses.
- */
+ */
-
+
-package io.moquette.spi.persistence;
+package io.moquette.spi.persistence;
-
+
-import io.moquette.server.Constants;
+import io.moquette.server.Constants;
-import io.moquette.spi.ClientSession;
+import io.moquette.spi.ClientSession;
-import io.moquette.spi.IMessagesStore;
+import io.moquette.spi.IMessagesStore;
-import io.moquette.spi.IMessagesStore.StoredMessage;
+import io.moquette.spi.IMessagesStore.StoredMessage;
-import io.moquette.spi.ISessionsStore;
+import io.moquette.spi.ISessionsStore;
-import io.moquette.spi.MessageGUID;
+import io.moquette.spi.MessageGUID;
-import io.moquette.spi.impl.Utils;
+import io.moquette.spi.impl.Utils;
-import io.moquette.spi.impl.subscriptions.Subscription;
+import io.moquette.spi.impl.subscriptions.Subscription;
-import io.moquette.spi.impl.subscriptions.Topic;
+import io.moquette.spi.impl.subscriptions.Topic;
-import io.moquette.spi.persistence.MapDBPersistentStore.PersistentSession;
+import io.moquette.spi.persistence.MapDBPersistentStore.PersistentSession;
-import org.slf4j.Logger;
+import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.slf4j.LoggerFactory;
-import java.util.*;
+import java.util.*;
-import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.ArrayBlockingQueue;
-import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.BlockingQueue;
-
+
-/**
+/**
- *
+ *
- * @author andrea
+ * @author andrea
- */
+ */
-public class MemorySessionStore implements ISessionsStore {
+public class MemorySessionStore implements ISessionsStore {
-
+
-    private static final Logger LOG = LoggerFactory.getLogger(MemorySessionStore.class);
+    private static final Logger LOG = LoggerFactory.getLogger(MemorySessionStore.class);
-
+
-    private Map<String, Map<Topic, Subscription>> m_persistentSubscriptions = new HashMap<>();
+    private Map<String, Map<Topic, Subscription>> m_persistentSubscriptions = new HashMap<>();
-
+
-    private Map<String, MapDBPersistentStore.PersistentSession> m_persistentSessions = new HashMap<>();
+    private Map<String, MapDBPersistentStore.PersistentSession> m_persistentSessions = new HashMap<>();
-
+
-    // maps clientID->[MessageId -> guid]
+    // maps clientID->[MessageId -> guid]
-    private Map<String, Map<Integer, MessageGUID>> m_inflightStore = new HashMap<>();
+    private Map<String, Map<Integer, MessageGUID>> m_inflightStore = new HashMap<>();
-    // private Map<String, Set<Integer>> m_inflightIDs = new HashMap<>();
+    // private Map<String, Set<Integer>> m_inflightIDs = new HashMap<>();
-    // maps clientID->BlockingQueue
+    // maps clientID->BlockingQueue
-    private Map<String, BlockingQueue<StoredMessage>> queues = new HashMap<>();
+    private Map<String, BlockingQueue<StoredMessage>> queues = new HashMap<>();
-    // maps clientID->[MessageId -> guid]
+    // maps clientID->[MessageId -> guid]
-    private Map<String, Map<Integer, MessageGUID>> m_secondPhaseStore = new HashMap<>();
+    private Map<String, Map<Integer, MessageGUID>> m_secondPhaseStore = new HashMap<>();
-
+
-    private Map<String, Map<Integer, MessageGUID>> m_messageToGuids;
+    private Map<String, Map<Integer, MessageGUID>> m_messageToGuids;
-    private final IMessagesStore m_messagesStore;
+    private final IMessagesStore m_messagesStore;
-
+
-    public MemorySessionStore(IMessagesStore messagesStore, Map<String, Map<Integer, MessageGUID>> messageToGuids) {
+    public MemorySessionStore(IMessagesStore messagesStore, Map<String, Map<Integer, MessageGUID>> messageToGuids) {
-        m_messageToGuids = messageToGuids;
+        m_messageToGuids = messageToGuids;
-        this.m_messagesStore = messagesStore;
+        this.m_messagesStore = messagesStore;
-    }
+    }
-
+
-    @Override
+    @Override
-    public void removeSubscription(Topic topic, String clientID) {
+    public void removeSubscription(Topic topic, String clientID) {
-        LOG.debug("removeSubscription topic filter: {} for clientID: {}", topic, clientID);
+        LOG.debug("removeSubscription topic filter: {} for clientID: {}", topic, clientID);
-        if (!m_persistentSubscriptions.containsKey(clientID)) {
+        if (!m_persistentSubscriptions.containsKey(clientID)) {
-            return;
+            return;
-        }
+        }
-        Map<Topic, Subscription> clientSubscriptions = m_persistentSubscriptions.get(clientID);
+        Map<Topic, Subscription> clientSubscriptions = m_persistentSubscriptions.get(clientID);
-        clientSubscriptions.remove(topic);
+        clientSubscriptions.remove(topic);
-    }
+    }
-
+
-    @Override
+    @Override
-    public void initStore() {
+    public void initStore() {
-
+
-    }
+    }
-
+
-    @Override
+    @Override
-    public void addNewSubscription(Subscription newSubscription) {
+    public void addNewSubscription(Subscription newSubscription) {
-        final String clientID = newSubscription.getClientId();
+        final String clientID = newSubscription.getClientId();
-        if (!m_persistentSubscriptions.containsKey(clientID)) {
+        if (!m_persistentSubscriptions.containsKey(clientID)) {
-            m_persistentSubscriptions.put(clientID, new HashMap<Topic, Subscription>());
+            m_persistentSubscriptions.put(clientID, new HashMap<Topic, Subscription>());
-        }
+        }
-
+
-        m_persistentSubscriptions.get(clientID).put(newSubscription.getTopicFilter(), newSubscription);
+        m_persistentSubscriptions.get(clientID).put(newSubscription.getTopicFilter(), newSubscription);
-    }
+    }
-
+
-    @Override
+    @Override
-    public void wipeSubscriptions(String clientID) {
+    public void wipeSubscriptions(String clientID) {
-        m_persistentSubscriptions.remove(clientID);
+        m_persistentSubscriptions.remove(clientID);
-    }
+    }
-
+
-    @Override
+    @Override
-    public boolean contains(String clientID) {
+    public boolean contains(String clientID) {
-        return m_persistentSubscriptions.containsKey(clientID);
+        return m_persistentSubscriptions.containsKey(clientID);
-    }
+    }
-
+
-    @Override
+    @Override
-    public ClientSession createNewSession(String clientID, boolean cleanSession) {
+    public ClientSession createNewSession(String clientID, boolean cleanSession) {
-        LOG.debug("createNewSession for client <{}>", clientID);
+        LOG.debug("createNewSession for client <{}>", clientID);
-        if (m_persistentSessions.containsKey(clientID)) {
+        if (m_persistentSessions.containsKey(clientID)) {
-            LOG.error("already exists a session for client <{}>, bad condition", 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);
+            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);
+        LOG.debug("clientID {} is a newcome, creating it's empty subscriptions set", clientID);
-        m_persistentSubscriptions.put(clientID, new HashMap<Topic, Subscription>());
+        m_persistentSubscriptions.put(clientID, new HashMap<Topic, Subscription>());
-        m_persistentSessions.put(clientID, new MapDBPersistentStore.PersistentSession(cleanSession));
+        m_persistentSessions.put(clientID, new MapDBPersistentStore.PersistentSession(cleanSession));
-        return new ClientSession(clientID, m_messagesStore, this, cleanSession);
+        return new ClientSession(clientID, m_messagesStore, this, cleanSession);
-    }
+    }
-
+
-    @Override
+    @Override
-    public ClientSession sessionForClient(String clientID) {
+    public ClientSession sessionForClient(String clientID) {
-        if (!m_persistentSessions.containsKey(clientID)) {
+        if (!m_persistentSessions.containsKey(clientID)) {
-            return null;
+            return null;
-        }
+        }
-
+
-        MapDBPersistentStore.PersistentSession storedSession = m_persistentSessions.get(clientID);
+        MapDBPersistentStore.PersistentSession storedSession = m_persistentSessions.get(clientID);
-        return new ClientSession(clientID, m_messagesStore, this, storedSession.cleanSession);
+        return new ClientSession(clientID, m_messagesStore, this, storedSession.cleanSession);
-    }
+    }
-
+
-    @Override
+    @Override
-    public Collection<ClientSession> getAllSessions() {
+    public Collection<ClientSession> getAllSessions() {
-        Collection<ClientSession> result = new ArrayList<>();
+        Collection<ClientSession> result = new ArrayList<>();
-        for (Map.Entry<String, PersistentSession> entry : m_persistentSessions.entrySet()) {
+        for (Map.Entry<String, PersistentSession> entry : m_persistentSessions.entrySet()) {
-            result.add(new ClientSession(entry.getKey(), m_messagesStore, this, entry.getValue().cleanSession));
+            result.add(new ClientSession(entry.getKey(), m_messagesStore, this, entry.getValue().cleanSession));
-        }
+        }
-        return result;
+        return result;
-    }
+    }
-
+
-    @Override
+    @Override
-    public void updateCleanStatus(String clientID, boolean cleanSession) {
+    public void updateCleanStatus(String clientID, boolean cleanSession) {
-        m_persistentSessions.put(clientID, new MapDBPersistentStore.PersistentSession(cleanSession));
+        m_persistentSessions.put(clientID, new MapDBPersistentStore.PersistentSession(cleanSession));
-    }
+    }
-
+
-    @Override
+    @Override
-    public List<ClientTopicCouple> listAllSubscriptions() {
+    public List<ClientTopicCouple> listAllSubscriptions() {
-        List<ClientTopicCouple> allSubscriptions = new ArrayList<>();
+        List<ClientTopicCouple> allSubscriptions = new ArrayList<>();
-        for (Map.Entry<String, Map<Topic, Subscription>> entry : m_persistentSubscriptions.entrySet()) {
+        for (Map.Entry<String, Map<Topic, Subscription>> entry : m_persistentSubscriptions.entrySet()) {
-            for (Subscription sub : entry.getValue().values()) {
+            for (Subscription sub : entry.getValue().values()) {
-                allSubscriptions.add(sub.asClientTopicCouple());
+                allSubscriptions.add(sub.asClientTopicCouple());
-            }
+            }
-        }
+        }
-        return allSubscriptions;
+        return allSubscriptions;
-    }
+    }
-
+
-    @Override
+    @Override
-    public Subscription getSubscription(ClientTopicCouple couple) {
+    public Subscription getSubscription(ClientTopicCouple couple) {
-        Map<Topic, Subscription> subscriptions = m_persistentSubscriptions.get(couple.clientID);
+        Map<Topic, Subscription> subscriptions = m_persistentSubscriptions.get(couple.clientID);
-        if (subscriptions == null || subscriptions.isEmpty()) {
+        if (subscriptions == null || subscriptions.isEmpty()) {
-            return null;
+            return null;
-        }
+        }
-        return subscriptions.get(couple.topicFilter);
+        return subscriptions.get(couple.topicFilter);
-    }
+    }
-
+
-    @Override
+    @Override
-    public List<Subscription> getSubscriptions() {
+    public List<Subscription> getSubscriptions() {
-        List<Subscription> subscriptions = new ArrayList<>();
+        List<Subscription> subscriptions = new ArrayList<>();
-        for (Map.Entry<String, Map<Topic, Subscription>> entry : m_persistentSubscriptions.entrySet()) {
+        for (Map.Entry<String, Map<Topic, Subscription>> entry : m_persistentSubscriptions.entrySet()) {
-            subscriptions.addAll(entry.getValue().values());
+            subscriptions.addAll(entry.getValue().values());
-        }
+        }
-        return subscriptions;
+        return subscriptions;
-    }
+    }
-
+
-    @Override
+    @Override
-    public void inFlightAck(String clientID, int messageID) {
+    public void inFlightAck(String clientID, int messageID) {
-        Map<Integer, MessageGUID> m = this.m_inflightStore.get(clientID);
+        Map<Integer, MessageGUID> m = this.m_inflightStore.get(clientID);
-        if (m == null) {
+        if (m == null) {
-            LOG.error("Can't find the inFlight record for client <{}>", clientID);
+            LOG.error("Can't find the inFlight record for client <{}>", clientID);
-            return;
+            return;
-        }
+        }
-        m.remove(messageID);
+        m.remove(messageID);
-    }
+    }
-
+
-    @Override
+    @Override
-    public void inFlight(String clientID, int messageID, MessageGUID guid) {
+    public void inFlight(String clientID, int messageID, MessageGUID guid) {
-        Map<Integer, MessageGUID> m = this.m_inflightStore.get(clientID);
+        Map<Integer, MessageGUID> m = this.m_inflightStore.get(clientID);
-        if (m == null) {
+        if (m == null) {
-            m = new HashMap<>();
+            m = new HashMap<>();
-        }
+        }
-        m.put(messageID, guid);
+        m.put(messageID, guid);
-        this.m_inflightStore.put(clientID, m);
+        this.m_inflightStore.put(clientID, m);
-    }
+    }
-
+
-    /**
+    /**
-     * Return the next valid packetIdentifier for the given client session.
+     * Return the next valid packetIdentifier for the given client session.
-     */
+     */
-    @Override
+    @Override
-    public int nextPacketID(String clientID) {
+    public int nextPacketID(String clientID) {
-        Map<Integer, MessageGUID> m = this.m_inflightStore.get(clientID);
+        Map<Integer, MessageGUID> m = this.m_inflightStore.get(clientID);
-        if (m == null) {
+        if (m == null) {
-            m = new HashMap<>();
+            m = new HashMap<>();
-            int nextPacketId = 1;
+            int nextPacketId = 1;
-            m.put(nextPacketId, null);
+            m.put(nextPacketId, null);
-            return nextPacketId;
+            return nextPacketId;
-        }
+        }
-        int maxId = Collections.max(m.keySet());
+        int maxId = Collections.max(m.keySet());
-        int nextPacketId = (maxId + 1) % 0xFFFF;
+        int nextPacketId = (maxId + 1) % 0xFFFF;
-        m.put(nextPacketId, null);
+        m.put(nextPacketId, null);
-        return nextPacketId;
+        return nextPacketId;
-    }
+    }
-
+
-    @Override
+    @Override
-    public BlockingQueue<StoredMessage> queue(String clientID) {
+    public BlockingQueue<StoredMessage> queue(String clientID) {
-        return Utils.defaultGet(queues, clientID, new ArrayBlockingQueue<StoredMessage>(Constants.MAX_MESSAGE_QUEUE));
+        return Utils.defaultGet(queues, clientID, new ArrayBlockingQueue<StoredMessage>(Constants.MAX_MESSAGE_QUEUE));
-    }
+    }
-
+
-    @Override
+    @Override
-    public void dropQueue(String clientID) {
+    public void dropQueue(String clientID) {
-        queues.remove(clientID);
+        queues.remove(clientID);
-    }
+    }
-
+
-    @Override
+    @Override
-    public void moveInFlightToSecondPhaseAckWaiting(String clientID, int messageID) {
+    public void moveInFlightToSecondPhaseAckWaiting(String clientID, int messageID) {
-        LOG.info("acknowledging inflight clientID <{}> messageID {}", clientID, messageID);
+        LOG.info("acknowledging inflight clientID <{}> messageID {}", clientID, messageID);
-        Map<Integer, MessageGUID> m = this.m_inflightStore.get(clientID);
+        Map<Integer, MessageGUID> m = this.m_inflightStore.get(clientID);
-        if (m == null) {
+        if (m == null) {
-            LOG.error("Can't find the inFlight record for client <{}>", clientID);
+            LOG.error("Can't find the inFlight record for client <{}>", clientID);
-            return;
+            return;
-        }
+        }
-        MessageGUID guid = m.remove(messageID);
+        MessageGUID guid = m.remove(messageID);
-
+
-        LOG.info("Moving to second phase store");
+        LOG.info("Moving to second phase store");
-        Map<Integer, MessageGUID> messageIDs = Utils
+        Map<Integer, MessageGUID> messageIDs = Utils
-                .defaultGet(m_secondPhaseStore, clientID, new HashMap<Integer, MessageGUID>());
+                .defaultGet(m_secondPhaseStore, clientID, new HashMap<Integer, MessageGUID>());
-        messageIDs.put(messageID, guid);
+        messageIDs.put(messageID, guid);
-        m_secondPhaseStore.put(clientID, messageIDs);
+        m_secondPhaseStore.put(clientID, messageIDs);
-    }
+    }
-
+
-    @Override
+    @Override
-    public MessageGUID secondPhaseAcknowledged(String clientID, int messageID) {
+    public MessageGUID secondPhaseAcknowledged(String clientID, int messageID) {
-        Map<Integer, MessageGUID> messageIDs = Utils
+        Map<Integer, MessageGUID> messageIDs = Utils
-                .defaultGet(m_secondPhaseStore, clientID, new HashMap<Integer, MessageGUID>());
+                .defaultGet(m_secondPhaseStore, clientID, new HashMap<Integer, MessageGUID>());
-        MessageGUID guid = messageIDs.remove(messageID);
+        MessageGUID guid = messageIDs.remove(messageID);
-        m_secondPhaseStore.put(clientID, messageIDs);
+        m_secondPhaseStore.put(clientID, messageIDs);
-        return guid;
+        return guid;
-    }
+    }
-
+
-    @Override
+    @Override
-    public MessageGUID mapToGuid(String clientID, int messageID) {
+    public MessageGUID mapToGuid(String clientID, int messageID) {
-        HashMap<Integer, MessageGUID> guids = (HashMap<Integer, MessageGUID>) Utils
+        HashMap<Integer, MessageGUID> guids = (HashMap<Integer, MessageGUID>) Utils
-                .defaultGet(m_messageToGuids, clientID, new HashMap<Integer, MessageGUID>());
+                .defaultGet(m_messageToGuids, clientID, new HashMap<Integer, MessageGUID>());
-        return guids.get(messageID);
+        return guids.get(messageID);
-    }
+    }
-
+
-    @Override
+    @Override
-    public StoredMessage getInflightMessage(String clientID, int messageID) {
+    public StoredMessage getInflightMessage(String clientID, int messageID) {
-        return null;
+        return null;
-    }
+    }
-
+
-    @Override
+    @Override
-    public int getInflightMessagesNo(String clientID) {
+    public int getInflightMessagesNo(String clientID) {
-        Map<Integer, MessageGUID> inflightMessages = m_inflightStore.get(clientID);
+        Map<Integer, MessageGUID> inflightMessages = m_inflightStore.get(clientID);
-        if (inflightMessages == null)
+        if (inflightMessages == null)
-            return 0;
+            return 0;
-        else
+        else
-            return inflightMessages.size();
+            return inflightMessages.size();
-    }
+    }
-
+
-    @Override
+    @Override
-    public int getPendingPublishMessagesNo(String clientID) {
+    public int getPendingPublishMessagesNo(String clientID) {
-        return m_messagesStore.getPendingPublishMessages(clientID);
+        return m_messagesStore.getPendingPublishMessages(clientID);
-    }
+    }
-
+
-    @Override
+    @Override
-    public int getSecondPhaseAckPendingMessages(String clientID) {
+    public int getSecondPhaseAckPendingMessages(String clientID) {
-        Map<Integer, MessageGUID> pendingAcks = m_secondPhaseStore.get(clientID);
+        Map<Integer, MessageGUID> pendingAcks = m_secondPhaseStore.get(clientID);
-        if (pendingAcks == null)
+        if (pendingAcks == null)
-            return 0;
+            return 0;
-        else
+        else
-            return pendingAcks.size();
+            return pendingAcks.size();
-    }
+    }
-
+
-}
+}
@@ -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
@@ -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
@@ -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");
    }
-}
+}