From 5c6889fbd5fc3457e37256d1694711aa52a6782c Mon Sep 17 00:00:00 2001 From: andsel Date: Sat, 25 Mar 2017 16:05:45 +0100 Subject: [PATCH] Separated persistence layer, created submodule for MapDB persistence letting the broker use the in-memory message store --- .gitignore | 3 + broker/pom.xml | 16 +- .../persistence}/MemoryMessagesStore.java | 20 ++- .../persistence/MemorySessionStore.java | 30 ++-- .../persistence}/MemoryStorageService.java | 3 +- .../persistence/PersistentSession.java | 32 ++++ .../netty/metrics/MQTTMessageLogger.java | 7 +- .../java/io/moquette/spi/ClientSession.java | 24 ++- .../spi/impl/InternalRepublisher.java | 49 ++---- .../moquette/spi/impl/MessagesPublisher.java | 8 +- .../impl/PersistentQueueMessageSender.java | 28 +-- .../moquette/spi/impl/ProtocolProcessor.java | 88 +++++----- .../impl/ProtocolProcessorBootstrapper.java | 141 +++++---------- .../moquette/spi/impl/Qos0PublishHandler.java | 18 +- .../moquette/spi/impl/Qos1PublishHandler.java | 37 ++-- .../moquette/spi/impl/Qos2PublishHandler.java | 44 ++--- .../spi/impl/subscriptions/Topic.java | 2 +- .../server/ConfigurationClassLoaderTest.java | 6 +- .../io/moquette/server/IntegrationUtils.java | 4 +- .../ServerIntegrationDBAuthenticatorTest.java | 4 - .../ServerIntegrationEmbeddedPublishTest.java | 4 - .../server/ServerIntegrationFuseTest.java | 1 - ...rationHazelcastHandlerInterceptorTest.java | 3 - .../server/ServerIntegrationPahoTest.java | 5 +- .../ServerIntegrationQoSValidationTest.java | 1 - .../ServerIntegrationWebSocketTest.java | 1 - ...erverLowlevelMessagesIntegrationTests.java | 20 +-- .../server/ServerRestartIntegrationTest.java | 3 - .../io/moquette/spi/ClientSessionTest.java | 2 +- .../spi/impl/ProtocolProcessorTest.java | 68 ++------ .../impl/ProtocolProcessor_CONNECT_Test.java | 1 + .../subscriptions/SubscriptionsStoreTest.java | 2 +- mapdb_storage/pom.xml | 164 ++++++++++++++++++ .../mapdb}/MapDBMessagesStore.java | 2 +- .../mapdb}/MapDBPersistentStore.java | 26 +-- .../mapdb}/MapDBSessionsStore.java | 85 ++++----- .../mapdb}/MapDBPersistentStoreTest.java | 20 ++- .../impl/subscriptions/PerformanceTest.java | 3 +- pom.xml | 1 + 39 files changed, 477 insertions(+), 499 deletions(-) rename broker/src/{test/java/io/moquette/spi/impl => main/java/io/moquette/persistence}/MemoryMessagesStore.java (82%) rename broker/src/{test/java/io/moquette/spi => main/java/io/moquette}/persistence/MemorySessionStore.java (87%) rename broker/src/{test/java/io/moquette/spi/impl => main/java/io/moquette/persistence}/MemoryStorageService.java (94%) create mode 100644 broker/src/main/java/io/moquette/persistence/PersistentSession.java create mode 100644 mapdb_storage/pom.xml rename {broker/src/main/java/io/moquette/spi/persistence => mapdb_storage/src/main/java/io/moquette/persistence/mapdb}/MapDBMessagesStore.java (99%) rename {broker/src/main/java/io/moquette/spi/persistence => mapdb_storage/src/main/java/io/moquette/persistence/mapdb}/MapDBPersistentStore.java (85%) rename {broker/src/main/java/io/moquette/spi/persistence => mapdb_storage/src/main/java/io/moquette/persistence/mapdb}/MapDBSessionsStore.java (77%) rename {broker/src/test/java/io/moquette/spi/persistence => mapdb_storage/src/test/java/io/moquette/persistence/mapdb}/MapDBPersistentStoreTest.java (91%) diff --git a/.gitignore b/.gitignore index b36831cb..8a3e0c20 100644 --- a/.gitignore +++ b/.gitignore @@ -16,6 +16,7 @@ broker/runner /bundle/runner/ /osgi_test/target/ /perf/target/ +/mapdb_storage/target/ *~ *log* .DS_Store @@ -40,3 +41,5 @@ broker/*moquette_store.* embedding_moquette/target/ broker/nb-configuration.xml broker/nbactions-Server.xml + +maven-metadata.xml diff --git a/broker/pom.xml b/broker/pom.xml index 2e7f9983..a0a256a6 100644 --- a/broker/pom.xml +++ b/broker/pom.xml @@ -18,6 +18,13 @@ + + + + + + + org.slf4j slf4j-api @@ -72,15 +79,6 @@ 1.10 - - org.mapdb - mapdb - - - - 1.0.8 - - org.hdrhistogram HdrHistogram diff --git a/broker/src/test/java/io/moquette/spi/impl/MemoryMessagesStore.java b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java similarity index 82% rename from broker/src/test/java/io/moquette/spi/impl/MemoryMessagesStore.java rename to broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java index 2921b8fe..da1812be 100644 --- a/broker/src/test/java/io/moquette/spi/impl/MemoryMessagesStore.java +++ b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java @@ -14,11 +14,12 @@ * You may elect to redistribute this code under either of these licenses. */ -package io.moquette.spi.impl; +package io.moquette.persistence; import io.moquette.spi.IMessagesStore; import io.moquette.spi.IMatchingCondition; import io.moquette.spi.MessageGUID; +import io.moquette.spi.impl.Utils; import io.moquette.spi.impl.subscriptions.Topic; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -71,11 +72,10 @@ public class MemoryMessagesStore implements IMessagesStore { MessageGUID guid = new MessageGUID(UUID.randomUUID().toString()); storedMessage.setGuid(guid); m_persistentMessageStore.put(guid, storedMessage); - HashMap guids = (HashMap) defaultGet( - m_messageToGuids, - storedMessage.getClientID(), - new HashMap()); + final HashMap emptyGuids = new HashMap<>(); + Map guids = defaultGet(m_messageToGuids, storedMessage.getClientID(), emptyGuids); guids.put(storedMessage.getMessageID(), guid); + m_messageToGuids.put(storedMessage.getClientID(), guids); return guid; } @@ -85,7 +85,11 @@ public class MemoryMessagesStore implements IMessagesStore { if (messageGUIDMap == null || messageGUIDMap.isEmpty()) { return; } - for (MessageGUID guid : messageGUIDMap.values()) { + //remove all guids from retained + Collection messagesToRemove = new HashSet<>(messageGUIDMap.values()); + messagesToRemove.removeAll(m_retainedStore.values()); + + for (MessageGUID guid : messagesToRemove) { m_persistentMessageStore.remove(guid); } } @@ -111,8 +115,8 @@ public class MemoryMessagesStore implements IMessagesStore { @Override public MessageGUID mapToGuid(String clientID, int messageID) { - HashMap guids = (HashMap) Utils - .defaultGet(m_messageToGuids, clientID, new HashMap()); + final HashMap emptyGuids = new HashMap<>(); + Map guids = Utils.defaultGet(m_messageToGuids, clientID, emptyGuids); return guids.get(messageID); } } diff --git a/broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java b/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java similarity index 87% rename from broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java rename to broker/src/main/java/io/moquette/persistence/MemorySessionStore.java index 61608aef..1af19c8f 100644 --- a/broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java +++ b/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java @@ -14,7 +14,7 @@ * You may elect to redistribute this code under either of these licenses. */ -package io.moquette.spi.persistence; +package io.moquette.persistence; import io.moquette.server.Constants; import io.moquette.spi.ClientSession; @@ -25,7 +25,6 @@ import io.moquette.spi.MessageGUID; import io.moquette.spi.impl.Utils; import io.moquette.spi.impl.subscriptions.Subscription; import io.moquette.spi.impl.subscriptions.Topic; -import io.moquette.spi.persistence.MapDBPersistentStore.PersistentSession; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.*; @@ -42,7 +41,7 @@ public class MemorySessionStore implements ISessionsStore { private Map> m_persistentSubscriptions = new HashMap<>(); - private Map m_persistentSessions = new HashMap<>(); + private Map m_persistentSessions = new HashMap<>(); // maps clientID->[MessageId -> guid] private Map> m_inflightStore = new HashMap<>(); @@ -102,7 +101,7 @@ public class MemorySessionStore implements ISessionsStore { } LOG.debug("clientID {} is a newcome, creating it's empty subscriptions set", clientID); m_persistentSubscriptions.put(clientID, new HashMap()); - m_persistentSessions.put(clientID, new MapDBPersistentStore.PersistentSession(cleanSession)); + m_persistentSessions.put(clientID, new PersistentSession(cleanSession)); return new ClientSession(clientID, m_messagesStore, this, cleanSession); } @@ -112,7 +111,7 @@ public class MemorySessionStore implements ISessionsStore { return null; } - MapDBPersistentStore.PersistentSession storedSession = m_persistentSessions.get(clientID); + PersistentSession storedSession = m_persistentSessions.get(clientID); return new ClientSession(clientID, m_messagesStore, this, storedSession.cleanSession); } @@ -127,7 +126,7 @@ public class MemorySessionStore implements ISessionsStore { @Override public void updateCleanStatus(String clientID, boolean cleanSession) { - m_persistentSessions.put(clientID, new MapDBPersistentStore.PersistentSession(cleanSession)); + m_persistentSessions.put(clientID, new PersistentSession(cleanSession)); } @Override @@ -191,7 +190,7 @@ public class MemorySessionStore implements ISessionsStore { m.put(nextPacketId, null); return nextPacketId; } - int maxId = Collections.max(m.keySet()); + int maxId = m.keySet().isEmpty() ? 0 :Collections.max(m.keySet()); int nextPacketId = (maxId + 1) % 0xFFFF; m.put(nextPacketId, null); return nextPacketId; @@ -199,7 +198,10 @@ public class MemorySessionStore implements ISessionsStore { @Override public BlockingQueue queue(String clientID) { - return Utils.defaultGet(queues, clientID, new ArrayBlockingQueue(Constants.MAX_MESSAGE_QUEUE)); + final ArrayBlockingQueue emptyQueue = new ArrayBlockingQueue<>(Constants.MAX_MESSAGE_QUEUE); + BlockingQueue messagesQueue = Utils.defaultGet(queues, clientID, emptyQueue); + queues.put(clientID, messagesQueue); + return messagesQueue; } @Override @@ -218,16 +220,16 @@ public class MemorySessionStore implements ISessionsStore { MessageGUID guid = m.remove(messageID); LOG.info("Moving to second phase store"); - Map messageIDs = Utils - .defaultGet(m_secondPhaseStore, clientID, new HashMap()); + final HashMap emptyGuids = new HashMap<>(); + Map messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, emptyGuids); messageIDs.put(messageID, guid); m_secondPhaseStore.put(clientID, messageIDs); } @Override public MessageGUID secondPhaseAcknowledged(String clientID, int messageID) { - Map messageIDs = Utils - .defaultGet(m_secondPhaseStore, clientID, new HashMap()); + final HashMap emptyGuids = new HashMap<>(); + Map messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, emptyGuids); MessageGUID guid = messageIDs.remove(messageID); m_secondPhaseStore.put(clientID, messageIDs); return guid; @@ -235,7 +237,9 @@ public class MemorySessionStore implements ISessionsStore { @Override public StoredMessage getInflightMessage(String clientID, int messageID) { - return null; + Map inflightMessages = m_inflightStore.get(clientID); + MessageGUID guid = inflightMessages.get(messageID); + return m_messagesStore.getMessageByGuid(guid); } @Override diff --git a/broker/src/test/java/io/moquette/spi/impl/MemoryStorageService.java b/broker/src/main/java/io/moquette/persistence/MemoryStorageService.java similarity index 94% rename from broker/src/test/java/io/moquette/spi/impl/MemoryStorageService.java rename to broker/src/main/java/io/moquette/persistence/MemoryStorageService.java index efb37241..225477cf 100644 --- a/broker/src/test/java/io/moquette/spi/impl/MemoryStorageService.java +++ b/broker/src/main/java/io/moquette/persistence/MemoryStorageService.java @@ -14,12 +14,11 @@ * You may elect to redistribute this code under either of these licenses. */ -package io.moquette.spi.impl; +package io.moquette.persistence; import io.moquette.spi.IMessagesStore; import io.moquette.spi.ISessionsStore; import io.moquette.spi.IStore; -import io.moquette.spi.persistence.MemorySessionStore; public class MemoryStorageService implements IStore { diff --git a/broker/src/main/java/io/moquette/persistence/PersistentSession.java b/broker/src/main/java/io/moquette/persistence/PersistentSession.java new file mode 100644 index 00000000..cab0ba21 --- /dev/null +++ b/broker/src/main/java/io/moquette/persistence/PersistentSession.java @@ -0,0 +1,32 @@ +/* + * Copyright (c) 2012-2017 The original author or authors + * ------------------------------------------------------ + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Apache License v2.0 which accompanies this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * The Apache License v2.0 is available at + * http://www.opensource.org/licenses/apache2.0.php + * + * You may elect to redistribute this code under either of these licenses. + */ +package io.moquette.persistence; + +import java.io.Serializable; + +/** + * This is a DTO used to persist minimal status (clean session and activation status) of a + * session. + */ +public class PersistentSession implements Serializable { + + private static final long serialVersionUID = 5052054783220481854L; + public final boolean cleanSession; + + public PersistentSession(boolean cleanSession) { + this.cleanSession = cleanSession; + } +} diff --git a/broker/src/main/java/io/moquette/server/netty/metrics/MQTTMessageLogger.java b/broker/src/main/java/io/moquette/server/netty/metrics/MQTTMessageLogger.java index c4edf7e4..bf9707b1 100644 --- a/broker/src/main/java/io/moquette/server/netty/metrics/MQTTMessageLogger.java +++ b/broker/src/main/java/io/moquette/server/netty/metrics/MQTTMessageLogger.java @@ -54,11 +54,8 @@ public class MQTTMessageLogger extends ChannelDuplexHandler { break; case SUBSCRIBE: MqttSubscribeMessage subscribe = (MqttSubscribeMessage) msg; - LOG.info( - "{} SUBSCRIBE <{}> to topics {}", - direction, - clientID, - subscribe.payload().topicSubscriptions()); + LOG.info("{} SUBSCRIBE <{}> to topics {}", direction, clientID, + subscribe.payload().topicSubscriptions()); break; case UNSUBSCRIBE: MqttUnsubscribeMessage unsubscribe = (MqttUnsubscribeMessage) msg; diff --git a/broker/src/main/java/io/moquette/spi/ClientSession.java b/broker/src/main/java/io/moquette/spi/ClientSession.java index c3f12ab0..663f6929 100644 --- a/broker/src/main/java/io/moquette/spi/ClientSession.java +++ b/broker/src/main/java/io/moquette/spi/ClientSession.java @@ -74,7 +74,7 @@ public class ClientSession { * @return the list of messages to be delivered for client related to the session. */ public BlockingQueue queue() { - LOG.info("Retrieving stored messages. MqttClientId = {}.", clientID); + LOG.info("Retrieving enqueued messages. ClientId={}", clientID); return this.m_sessionsStore.queue(clientID); } @@ -85,14 +85,14 @@ public class ClientSession { public boolean subscribe(Subscription newSubscription) { LOG.info( - "Adding new subscription. MqttClientId = {}, topics = {}, qos = {}.", + "Adding new subscription. ClientId={}, topics={}, qos={}", newSubscription.getClientId(), newSubscription.getTopicFilter(), newSubscription.getRequestedQos()); boolean validTopic = newSubscription.getTopicFilter().isValid(); if (!validTopic) { LOG.warn( - "The topic filter is not valid. MqttClientId = {}, topics = {}.", + "The topic filter is not valid. ClientId={}, topics={}", newSubscription.getClientId(), newSubscription.getTopicFilter()); // send SUBACK with 0x80 for this topic filter @@ -103,13 +103,9 @@ public class ClientSession { // update the selected subscriptions if not present or if has a greater qos if (existingSub == null || existingSub.getRequestedQos().value() < newSubscription.getRequestedQos().value()) { if (existingSub != null) { - LOG.info( - "The subscription already existed with a lower QoS value. It will be updated. " - + "MqttClientId = {}, topics = {}, existingQos = {}, newQos = {}.", - newSubscription.getClientId(), - newSubscription.getTopicFilter(), - existingSub.getRequestedQos(), - newSubscription.getRequestedQos()); + LOG.info("Subscription already existed with a lower QoS value. It will be updated. ClientId={}, " + + "topics={}, existingQos={}, newQos={}", newSubscription.getClientId(), + newSubscription.getTopicFilter(), existingSub.getRequestedQos(), newSubscription.getRequestedQos()); subscriptions.remove(newSubscription); } subscriptions.add(newSubscription); @@ -119,7 +115,7 @@ public class ClientSession { } public void unsubscribeFrom(Topic topicFilter) { - LOG.info("Removing subscription. MqttClientID = {}, topics = {}.", clientID, topicFilter); + LOG.info("Removing subscription. ClientID={}, topics={}", clientID, topicFilter); m_sessionsStore.removeSubscription(topicFilter, clientID); Set subscriptionsToRemove = new HashSet<>(); for (Subscription sub : this.subscriptions) { @@ -132,18 +128,18 @@ public class ClientSession { public void disconnect() { if (this.cleanSession) { - LOG.info("The client has disconnected. Removing its subscriptions. MqttClientId = {}.", clientID); + LOG.info("Client disconnected. Removing its subscriptions. ClientId={}", clientID); // cleanup topic subscriptions cleanSession(); } } public void cleanSession() { - LOG.info("Wiping existing subscriptions. MqttClientId = {}.", this.clientID); + LOG.info("Wiping existing subscriptions. ClientId={}", this.clientID); m_sessionsStore.wipeSubscriptions(this.clientID); // remove also the messages stored of type QoS1/2 - LOG.info("Removing stored messages with QoS 1 and 2. MqttClientId = {}.", this.clientID); + LOG.info("Removing stored messages with QoS 1 and 2. ClientId={}", this.clientID); messagesStore.dropMessagesInSession(this.clientID); } diff --git a/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java b/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java index 7db968a0..5206c5ba 100644 --- a/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java +++ b/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java @@ -27,13 +27,8 @@ class InternalRepublisher { // fire as retained the message Integer packetID = storedMsg.getQos() == MqttQoS.AT_MOST_ONCE ? null : targetSession.nextPacketId(); if (packetID != null) { - LOG.debug( - "Adding message to inflight zone. " - + "MqttClientId = {}, packetId = {}, messageId = {}, topic = {}.", - targetSession.clientID, - packetID, - storedMsg.getMessageID(), - storedMsg.getTopic()); + LOG.debug("Adding message to inflight zone. ClientId={}, packetId={}, messageId={}, topic={}", + targetSession.clientID, packetID, storedMsg.getMessageID(), storedMsg.getTopic()); targetSession.inFlightAckWaiting(storedMsg.getGuid(), packetID); } MqttPublishMessage publishMsg = retainedPublish(storedMsg); @@ -51,12 +46,8 @@ class InternalRepublisher { for (IMessagesStore.StoredMessage pubEvt : storedPublishes) { // put in flight zone - LOG.debug( - "Adding message ot inflight zone. MqttClientId = {}, guid = {}, messageId = {}, topic = {}.", - clientSession.clientID, - pubEvt.getGuid(), - pubEvt.getMessageID(), - pubEvt.getTopic()); + LOG.debug("Adding message ot inflight zone. ClientId={}, guid={}, messageId={}, topic={}", + clientSession.clientID, pubEvt.getGuid(), pubEvt.getMessageID(), pubEvt.getTopic()); clientSession.inFlightAckWaiting(pubEvt.getGuid(), pubEvt.getMessageID()); MqttPublishMessage publishMsg = notRetainedPublish(pubEvt); // set the PacketIdentifier only for QoS > 0 @@ -68,39 +59,23 @@ class InternalRepublisher { } private MqttPublishMessage notRetainedPublish(IMessagesStore.StoredMessage storedMessage, Integer messageID) { - return createPublishForQos( - storedMessage.getTopic(), - storedMessage.getQos(), - storedMessage.getMessage(), - false, - messageID); + return createPublishForQos(storedMessage.getTopic(), storedMessage.getQos(), storedMessage.getMessage(), false, + messageID); } private MqttPublishMessage notRetainedPublish(IMessagesStore.StoredMessage storedMessage) { - return createPublishForQos( - storedMessage.getTopic(), - storedMessage.getQos(), - storedMessage.getMessage(), - false, - 0); + return createPublishForQos(storedMessage.getTopic(), storedMessage.getQos(), storedMessage.getMessage(), false, + 0); } private MqttPublishMessage retainedPublish(IMessagesStore.StoredMessage storedMessage) { - return createPublishForQos( - storedMessage.getTopic(), - storedMessage.getQos(), - storedMessage.getMessage(), - true, - 0); + return createPublishForQos(storedMessage.getTopic(), storedMessage.getQos(), storedMessage.getMessage(), true, + 0); } private MqttPublishMessage retainedPublish(IMessagesStore.StoredMessage storedMessage, Integer packetID) { - return createPublishForQos( - storedMessage.getTopic(), - storedMessage.getQos(), - storedMessage.getMessage(), - true, - packetID); + return createPublishForQos(storedMessage.getTopic(), storedMessage.getQos(), storedMessage.getMessage(), true, + packetID); } public static MqttPublishMessage createPublishForQos(String topic, MqttQoS qos, ByteBuf message, boolean retained, diff --git a/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java b/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java index 7d03cb36..e03a3e9d 100644 --- a/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java +++ b/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java @@ -80,12 +80,8 @@ class MessagesPublisher { this.messageSender.sendPublish(targetSession, publishMsg); } else { if (!targetSession.isCleanSession()) { - LOG.debug( - "Storing pending PUBLISH inactive message. MessageId={}, CId={}, topicFilter={}, qos={}", - pubMsg.getMessageID(), - sub.getClientId(), - sub.getTopicFilter(), - qos); + LOG.debug("Storing pending PUBLISH inactive message. MessageId={}, CId={}, topicFilter={}, qos={}", + pubMsg.getMessageID(), sub.getClientId(), sub.getTopicFilter(), qos); // store the message in targetSession queue to deliver targetSession.enqueue(pubMsg); } diff --git a/broker/src/main/java/io/moquette/spi/impl/PersistentQueueMessageSender.java b/broker/src/main/java/io/moquette/spi/impl/PersistentQueueMessageSender.java index 3dbef971..9afaa3f8 100644 --- a/broker/src/main/java/io/moquette/spi/impl/PersistentQueueMessageSender.java +++ b/broker/src/main/java/io/moquette/spi/impl/PersistentQueueMessageSender.java @@ -24,12 +24,8 @@ class PersistentQueueMessageSender { final int messageId = pubMessage.variableHeader().messageId(); final String topicName = pubMessage.variableHeader().topicName(); if (LOG.isDebugEnabled()) { - LOG.debug( - "Sending PUBLISH message. MessageId={}, CId={}, topic={}, qos={}, payload={}", - messageId, - clientId, - topicName, - DebugUtils.payload2Str(pubMessage.payload())); + LOG.debug("Sending PUBLISH message. MessageId={}, CId={}, topic={}, qos={}, payload={}", messageId, + clientId, topicName, DebugUtils.payload2Str(pubMessage.payload())); } else { LOG.info("Sending PUBLISH message. MessageId={}, CId={}, topic={}", messageId, clientId, topicName); } @@ -38,24 +34,12 @@ class PersistentQueueMessageSender { MqttQoS qos = pubMessage.fixedHeader().qosLevel(); if (!messageDelivered && qos != AT_MOST_ONCE && !clientsession.isCleanSession()) { - LOG.warn( - "The PUBLISH message could not be delivered. It will be stored. " - + "MessageId = {}, CId={}, topic = {}, qos = {}, cleanSession = {}.", - messageId, - clientId, - topicName, - qos, - false); + LOG.warn("PUBLISH message could not be delivered. It will be stored. MessageId={}, CId={}, topic={}, qos={}, " + + "cleanSession={}", messageId, clientId, topicName, qos, false); clientsession.enqueue(asStoredMessage(pubMessage)); } else { - LOG.warn( - "The PUBLISH message could not be delivered. It will be discarded. " - + "MessageId = {}, CId={}, topic = {}, qos = {}, cleanSession = {}.", - messageId, - clientId, - topicName, - qos, - true); + LOG.warn("PUBLISH message could not be delivered. It will be discarded. MessageId={}, CId={}, topic={}, " + + "qos={}, cleanSession={}", messageId, clientId, topicName, qos, true); } } } diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java index d0239bd4..c5b52510 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -209,7 +209,7 @@ public class ProtocolProcessor { this.allowZeroByteClientId = allowZeroByteClientId; m_authorizator = authorizator; if (LOG.isDebugEnabled()) { - LOG.debug("Initial subscriptions tree={}.", subscriptions.dumpTree()); + LOG.debug("Initial subscriptions tree={}", subscriptions.dumpTree()); } m_authenticator = authenticator; m_messagesStore = storageService; @@ -302,7 +302,7 @@ public class ProtocolProcessor { channel.close(); } - LOG.info("The CONNECT message has been processed. CId={}, username={}.", clientId, payload.userName()); + LOG.info("The CONNECT message has been processed. CId={}, username={}", clientId, payload.userName()); } private MqttConnAckMessage connAck(MqttConnectReturnCode returnCode) { @@ -347,7 +347,7 @@ public class ProtocolProcessor { } private boolean sendAck(ConnectionDescriptor descriptor, MqttConnectMessage msg, final String clientId) { - LOG.info("Sending connect ACK. CId={}.", clientId); + LOG.info("Sending connect ACK. CId={}", clientId); final boolean success = descriptor.assignState(DISCONNECTED, SENDACK); if (!success) { return false; @@ -363,17 +363,17 @@ public class ProtocolProcessor { } if (isSessionAlreadyStored) { - LOG.info("Cleaning session. CId={}.", clientId); + LOG.info("Cleaning session. CId={}", clientId); clientSession.cleanSession(msg.variableHeader().isCleanSession()); } descriptor.writeAndFlush(okResp); - LOG.info("The connect ACK has been sent. CId={}.", clientId); + LOG.info("The connect ACK has been sent. CId={}", clientId); return true; } private void initializeKeepAliveTimeout(Channel channel, MqttConnectMessage msg, final String clientId) { int keepAlive = msg.variableHeader().keepAliveTimeSeconds(); - LOG.info("Configuring connection. CId={}.", clientId); + LOG.info("Configuring connection. CId={}", clientId); NettyUtils.keepAlive(channel, keepAlive); // session.attr(NettyUtils.ATTR_KEY_CLEANSESSION).set(msg.variableHeader().isCleanSession()); NettyUtils.cleanSession(channel, msg.variableHeader().isCleanSession()); @@ -399,7 +399,7 @@ public class ProtocolProcessor { WillMessage will = new WillMessage(msg.payload().willTopic(), bb, msg.variableHeader().isWillRetain(), willQos); m_willStore.put(clientId, will); - LOG.info("MQTT last will and testament has been configured. CId={}.", clientId); + LOG.info("MQTT last will and testament has been configured. CId={}", clientId); } } @@ -416,7 +416,7 @@ public class ProtocolProcessor { clientSession = m_sessionsStore.createNewSession(clientId, msg.variableHeader().isCleanSession()); } if (msg.variableHeader().isCleanSession()) { - LOG.info("Cleaning session. CId={}.", clientId); + LOG.info("Cleaning session. CId={}", clientId); clientSession.cleanSession(); } return clientSession; @@ -453,10 +453,10 @@ public class ProtocolProcessor { * Republish QoS1 and QoS2 messages stored into the session for the clientID. */ private void republishStoredInSession(ClientSession clientSession) { - LOG.info("Republishing stored publish events. CId={}.", clientSession.clientID); + LOG.info("Republishing stored publish events. CId={}", clientSession.clientID); BlockingQueue publishedEvents = clientSession.queue(); if (publishedEvents.isEmpty()) { - LOG.info("There are no stored publish events. ClientId={}.", clientSession.clientID); + LOG.info("There are no stored publish events to ClientId={}", clientSession.clientID); return; } @@ -531,7 +531,7 @@ public class ProtocolProcessor { public void internalPublish(MqttPublishMessage msg, final String clientId) { final MqttQoS qos = msg.fixedHeader().qosLevel(); final Topic topic = new Topic(msg.variableHeader().topicName()); - LOG.info("Sending PUBLISH message. Topic={}, qos={}.", topic, qos); + LOG.info("Sending PUBLISH message. Topic={}, qos={}", topic, qos); MessageGUID guid = null; IMessagesStore.StoredMessage toStoreMsg = asStoredMessage(msg); @@ -579,7 +579,7 @@ public class ProtocolProcessor { Topic topic = new Topic(tobeStored.getTopic()); List topicMatchingSubscriptions = subscriptions.matches(topic); - LOG.info("Publishing will message. CId={}, messageId={}, topic={}.", clientID, messageId, topic); + LOG.info("Publishing will message. CId={}, messageId={}, topic={}", clientID, messageId, topic); this.messagesPublisher.publish2Subscribers(tobeStored, topicMatchingSubscriptions); } @@ -610,7 +610,7 @@ public class ProtocolProcessor { int messageID = messageId(msg); targetSession.moveInFlightToSecondPhaseAckWaiting(messageID); // once received a PUBREC reply with a PUBREL(messageID) - LOG.debug("Processing PUBREC message. CId={}, messageId={}.", clientID, messageID); + LOG.debug("Processing PUBREC message. CId={}, messageId={}", clientID, messageID); MqttFixedHeader pubRelHeader = new MqttFixedHeader(MqttMessageType.PUBREL, false, AT_LEAST_ONCE, false, 0); MqttMessage pubRelMessage = new MqttMessage(pubRelHeader, from(messageID)); @@ -620,7 +620,7 @@ public class ProtocolProcessor { public void processPubComp(Channel channel, MqttMessage msg) { String clientID = NettyUtils.clientID(channel); int messageID = messageId(msg); - LOG.debug("Processing PUBCOMP message. CId={}, messageId={}.", clientID, messageID); + LOG.debug("Processing PUBCOMP message. CId={}, messageId={}", clientID, messageID); // once received the PUBCOMP then remove the message from the temp memory ClientSession targetSession = m_sessionsStore.sessionForClient(clientID); StoredMessage inflightMsg = targetSession.secondPhaseAcknowledged(messageID); @@ -631,7 +631,7 @@ public class ProtocolProcessor { public void processDisconnect(Channel channel) throws InterruptedException { final String clientID = NettyUtils.clientID(channel); - LOG.info("Processing DISCONNECT message. CId={}.", clientID); + LOG.info("Processing DISCONNECT message. CId={}", clientID); channel.flush(); final ConnectionDescriptor existingDescriptor = this.connectionDescriptors.getConnection(clientID); if (existingDescriptor == null) { @@ -642,42 +642,42 @@ public class ProtocolProcessor { if (existingDescriptor.doesNotUseChannel(channel)) { // another client saved it's descriptor, exit - LOG.warn("Another client is using the connection descriptor. Closing connection. CId={}.", clientID); + LOG.warn("Another client is using the connection descriptor. Closing connection. CId={}", clientID); existingDescriptor.abort(); return; } if (!removeSubscriptions(existingDescriptor, clientID)) { - LOG.warn("Unable to remove subscriptions. Closing connection. CId={}.", clientID); + LOG.warn("Unable to remove subscriptions. Closing connection. CId={}", clientID); existingDescriptor.abort(); return; } if (!dropStoredMessages(existingDescriptor, clientID)) { - LOG.warn("Unable to drop stored messages. Closing connection. CId={}.", clientID); + LOG.warn("Unable to drop stored messages. Closing connection. CId={}", clientID); existingDescriptor.abort(); return; } if (!cleanWillMessageAndNotifyInterceptor(existingDescriptor, clientID)) { - LOG.warn("Unable to drop will message. Closing connection. CId={}.", clientID); + LOG.warn("Unable to drop will message. Closing connection. CId={}", clientID); existingDescriptor.abort(); return; } if (!existingDescriptor.close()) { - LOG.info("The connection has been closed. CId={}.", clientID); + LOG.info("The connection has been closed. CId={}", clientID); return; } boolean stillPresent = this.connectionDescriptors.removeConnection(existingDescriptor); if (!stillPresent) { // another descriptor was inserted - LOG.warn("Another descriptor has been inserted. CId={}.", clientID); + LOG.warn("Another descriptor has been inserted. CId={}", clientID); return; } - LOG.info("The DISCONNECT message has been processed. CId={}.", clientID); + LOG.info("The DISCONNECT message has been processed. CId={}", clientID); } private boolean removeSubscriptions(ConnectionDescriptor descriptor, String clientID) { @@ -687,9 +687,9 @@ public class ProtocolProcessor { } if (descriptor.cleanSession) { - LOG.info("Removing saved subscriptions. CId={}.", descriptor.clientID); + LOG.info("Removing saved subscriptions. CId={}", descriptor.clientID); m_sessionsStore.wipeSubscriptions(clientID); - LOG.info("The saved subscriptions have been removed. CId={}.", descriptor.clientID); + LOG.info("The saved subscriptions have been removed. CId={}", descriptor.clientID); } return true; } @@ -701,9 +701,9 @@ public class ProtocolProcessor { } if (descriptor.cleanSession) { - LOG.debug("Removing messages of session. CId={}.", descriptor.clientID); + LOG.debug("Removing messages of session. CId={}", descriptor.clientID); this.m_sessionsStore.dropQueue(clientID); - LOG.debug("The messages of the session have been removed. CId={}.", descriptor.clientID); + LOG.debug("The messages of the session have been removed. CId={}", descriptor.clientID); } return true; } @@ -714,7 +714,7 @@ public class ProtocolProcessor { return false; } - LOG.info("Removing will message. ClientId={}.", descriptor.clientID); + LOG.info("Removing will message. ClientId={}", descriptor.clientID); // cleanup the will store m_willStore.remove(clientID); String username = descriptor.getUsername(); @@ -723,7 +723,7 @@ public class ProtocolProcessor { } public void processConnectionLost(String clientID, Channel channel) { - LOG.info("Processing connection lost event. CId={}.", clientID); + LOG.info("Processing connection lost event. CId={}", clientID); ConnectionDescriptor oldConnDescr = new ConnectionDescriptor(clientID, channel, true); connectionDescriptors.removeConnection(oldConnDescr); // publish the Will message (if any) for the clientID @@ -750,7 +750,7 @@ public class ProtocolProcessor { List topics = msg.payload().topics(); String clientID = NettyUtils.clientID(channel); - LOG.info("Processing UNSUBSCRIBE message. CId={}, topics={}.", clientID, topics); + LOG.info("Processing UNSUBSCRIBE message. CId={}, topics={}", clientID, topics); ClientSession clientSession = m_sessionsStore.sessionForClient(clientID); for (String t : topics) { @@ -764,7 +764,7 @@ public class ProtocolProcessor { return; } - LOG.debug("Removing subscription. CId={}, topic={}.", clientID, topic); + LOG.debug("Removing subscription. CId={}, topic={}", clientID, topic); subscriptions.removeSubscription(topic, clientID); clientSession.unsubscribeFrom(topic); String username = NettyUtils.userName(channel); @@ -776,29 +776,28 @@ public class ProtocolProcessor { MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.UNSUBACK, false, AT_LEAST_ONCE, false, 0); MqttUnsubAckMessage ackMessage = new MqttUnsubAckMessage(fixedHeader, from(messageID)); - LOG.info("Sending UNSUBACK message. CId={}, topics={}, messageId={}.", clientID, topics, messageID); + LOG.info("Sending UNSUBACK message. CId={}, topics={}, messageId={}", clientID, topics, messageID); channel.writeAndFlush(ackMessage); } public void processSubscribe(Channel channel, MqttSubscribeMessage msg) { String clientID = NettyUtils.clientID(channel); int messageID = messageId(msg); - LOG.info("Processing SUBSCRIBE message. CId={}, messageId={}.", clientID, messageID); + LOG.info("Processing SUBSCRIBE message. CId={}, messageId={}", clientID, messageID); RunningSubscription executionKey = new RunningSubscription(clientID, messageID); - SubscriptionState currentStatus = this.subscriptionInCourse - .putIfAbsent(executionKey, SubscriptionState.VERIFIED); + SubscriptionState currentStatus = subscriptionInCourse.putIfAbsent(executionKey, SubscriptionState.VERIFIED); if (currentStatus != null) { - LOG.warn("Client sent another SUBSCRIBE message while this one was being processed " + - "CId={}, messageId={}", clientID, messageID); + LOG.warn("Client sent another SUBSCRIBE message while this one was being processed CId={}, messageId={}", + clientID, messageID); return; } String username = NettyUtils.userName(channel); List ackTopics = doVerify(clientID, username, msg); MqttSubAckMessage ackMessage = doAckMessageFromValidateFilters(ackTopics, messageID); if (!this.subscriptionInCourse.replace(executionKey, SubscriptionState.VERIFIED, SubscriptionState.STORED)) { - LOG.warn("Client sent another SUBSCRIBE message while the topic filters were being verified " + - "CId={}, messageId={}", clientID, messageID); + LOG.warn("Client sent another SUBSCRIBE message while the topic filters were being verified CId={}, " + + "messageId={}", clientID, messageID); return; } @@ -861,19 +860,18 @@ public class ProtocolProcessor { Topic topic = new Topic(req.topicName()); if (!m_authorizator.canRead(topic, username, clientSession.clientID)) { // send SUBACK with 0x80, the user hasn't credentials to read the topic - LOG.error("Client does not have read permissions on the topic " + - "CId={}, username={}, messageId={}, topic={}", - clientID, username, messageId, topic); + LOG.error("Client does not have read permissions on the topic CId={}, username={}, messageId={}, " + + "topic={}", clientID, username, messageId, topic); ackTopics.add(new MqttTopicSubscription(topic.toString(), FAILURE)); } else { MqttQoS qos; if (topic.isValid()) { - LOG.info("Client will be subscribed to the topic " + - "CId={}, username={}, messageId={}, topic={}", clientID, username, messageId, topic); + LOG.info("Client will be subscribed to the topic CId={}, username={}, messageId={}, topic={}", + clientID, username, messageId, topic); qos = req.qualityOfService(); } else { - LOG.error("Topic filter is not valid CId={}, username={}, messageId={}, topic={}", - clientID, username, messageId, topic); + LOG.error("Topic filter is not valid CId={}, username={}, messageId={}, topic={}", clientID, + username, messageId, topic); qos = FAILURE; } ackTopics.add(new MqttTopicSubscription(topic.toString(), qos)); diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java index d79cfd68..309de6c2 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java @@ -32,7 +32,6 @@ import io.moquette.spi.impl.security.PermitAllAuthorizator; import io.moquette.spi.impl.security.ResourceAuthenticator; import io.moquette.spi.impl.subscriptions.Subscription; import io.moquette.spi.impl.subscriptions.SubscriptionsStore; -import io.moquette.spi.persistence.MapDBPersistentStore; import io.moquette.spi.security.IAuthenticator; import io.moquette.spi.security.IAuthorizator; import org.slf4j.Logger; @@ -51,6 +50,7 @@ import java.util.List; public class ProtocolProcessorBootstrapper { private static final Logger LOG = LoggerFactory.getLogger(ProtocolProcessorBootstrapper.class); + public static final String MAPDB_STORE_CLASS = "io.moquette.persistence.MemoryStorageService"; private SubscriptionsStore subscriptions; @@ -83,7 +83,7 @@ public class ProtocolProcessorBootstrapper { * an implementation of the authorizator to be used, if null load that specified in * config and fallback on the default one (permit all). * @param server - * the serber to init. + * the server to init. * @return the processor created for the broker. */ public ProtocolProcessor init(IConfig props, List embeddedObservers, @@ -92,31 +92,21 @@ public class ProtocolProcessorBootstrapper { IMessagesStore messagesStore; LOG.info("Initializing messages and sessions stores..."); - String storageClassName = props.getProperty(BrokerConstants.STORAGE_CLASS_NAME, ""); - if (storageClassName != null && !storageClassName.isEmpty()) { - final IStore store = loadClass(storageClassName, IStore.class, Server.class, server); - messagesStore = store.messagesStore(); - m_sessionsStore = store.sessionsStore(); - storeShutdown = new Runnable() { - - @Override - public void run() { - store.close(); - } - }; - } else { - final MapDBPersistentStore m_mapStorage = new MapDBPersistentStore(props, server.getScheduler()); - m_mapStorage.initStore(); - messagesStore = m_mapStorage.messagesStore(); - m_sessionsStore = m_mapStorage.sessionsStore(); - storeShutdown = new Runnable() { - - @Override - public void run() { - m_mapStorage.close(); - } - }; + String storageClassName = props.getProperty(BrokerConstants.STORAGE_CLASS_NAME, MAPDB_STORE_CLASS); + if (storageClassName == null || storageClassName.isEmpty()) { + LOG.error("storage_class property not defined"); + throw new IllegalArgumentException("Can't start a valid persistence layer"); } + final IStore store = loadClass(storageClassName, IStore.class, Server.class, server); + messagesStore = store.messagesStore(); + m_sessionsStore = store.sessionsStore(); + storeShutdown = new Runnable() { + + @Override + public void run() { + store.close(); + } + }; LOG.info("Configuring message interceptors..."); @@ -148,7 +138,7 @@ public class ProtocolProcessorBootstrapper { } else { authenticator = new ResourceAuthenticator(resourceLoader, passwdPath); } - LOG.info("An {} authenticator instance will be used.", authenticator.getClass().getName()); + LOG.info("An {} authenticator instance will be used", authenticator.getClass().getName()); } LOG.info("Configuring MQTT authorizator..."); @@ -162,119 +152,78 @@ public class ProtocolProcessorBootstrapper { if (aclFilePath != null && !aclFilePath.isEmpty()) { authorizator = new DenyAllAuthorizator(); try { - LOG.info("Parsing ACL file. Path = {}.", aclFilePath); + LOG.info("Parsing ACL file. Path = {}", aclFilePath); authorizator = ACLFileParser.parse(resourceLoader.loadResource(aclFilePath)); } catch (ParseException pex) { - LOG.error( - "Unable to parse ACL file. Path = {}, cause = {}, errorMessage = {}.", - aclFilePath, - pex.getCause(), - pex.getMessage()); + LOG.error("Unable to parse ACL file. path="+ aclFilePath, pex); } } else { authorizator = new PermitAllAuthorizator(); } - LOG.info("An {} authorizator instance will be used.", authorizator.getClass().getName()); + LOG.info("An {} authorizator instance will be used", authorizator.getClass().getName()); } LOG.info("Initializing connection descriptor store..."); connectionDescriptors = new ConnectionDescriptorStore(m_sessionsStore); LOG.info("Initializing MQTT protocol processor..."); - boolean allowAnonymous = Boolean - .parseBoolean(props.getProperty(BrokerConstants.ALLOW_ANONYMOUS_PROPERTY_NAME, "true")); - boolean allowZeroByteClientId = Boolean - .parseBoolean(props.getProperty(BrokerConstants.ALLOW_ZERO_BYTE_CLIENT_ID_PROPERTY_NAME, "false")); - m_processor.init( - connectionDescriptors, - subscriptions, - messagesStore, - m_sessionsStore, - authenticator, - allowAnonymous, - allowZeroByteClientId, - authorizator, - m_interceptor, + boolean allowAnonymous = Boolean.parseBoolean(props.getProperty(BrokerConstants.ALLOW_ANONYMOUS_PROPERTY_NAME, "true")); + boolean allowZeroByteClientId = Boolean.parseBoolean(props.getProperty(BrokerConstants.ALLOW_ZERO_BYTE_CLIENT_ID_PROPERTY_NAME, "false")); + m_processor.init(connectionDescriptors, subscriptions, messagesStore, m_sessionsStore, authenticator, + allowAnonymous, allowZeroByteClientId, authorizator, m_interceptor, props.getProperty(BrokerConstants.PORT_PROPERTY_NAME)); return m_processor; } @SuppressWarnings("unchecked") - private T loadClass(String className, Class iface, Class constructorArgClass, U props) { + private T loadClass(String className, Class intrface, Class constructorArgClass, U props) { T instance = null; try { - LOG.info("Loading class. ClassName = {}, interfaceName = {}.", className, iface.getName()); + LOG.info("Loading class. ClassName={}, interfaceName={}", className, intrface.getName()); Class clazz = Class.forName(className); // check if method getInstance exists Method method = clazz.getMethod("getInstance", new Class[]{}); try { - LOG.info( - "Invoking getInstance() method. ClassName = {}, interfaceName = {}.", - className, - iface.getName()); + LOG.info("Invoking getInstance() method. ClassName = {}, interfaceName = {}.", className, intrface.getName()); instance = (T) method.invoke(null, new Object[]{}); } catch (IllegalArgumentException | InvocationTargetException | IllegalAccessException ex) { - LOG.error( - "Unable to invoke getInstance() method. " - + "ClassName = {}, interfaceName = {}, cause = {}, errorMessage = {}.", - className, - iface.getName(), - ex.getCause(), - ex.getMessage()); + LOG.error("Unable to invoke getInstance() method. ClassName = {}, interfaceName = {}, cause = {}, errorMessage = {}.", + className, intrface.getName(), ex.getCause(), ex.getMessage()); return null; } } catch (NoSuchMethodException nsmex) { try { // check if constructor with constructor arg class parameter // exists - LOG.info( - "Invoking constructor with {} argument. ClassName = {}, interfaceName = {}.", - constructorArgClass.getName(), - className, - iface.getName()); - instance = this.getClass().getClassLoader().loadClass(className).asSubclass(iface) - .getConstructor(constructorArgClass).newInstance(props); + LOG.info("Invoking constructor with {} argument. ClassName = {}, interfaceName = {}.", + constructorArgClass.getName(), className, intrface.getName()); + instance = this.getClass().getClassLoader() + .loadClass(className) + .asSubclass(intrface) + .getConstructor(constructorArgClass) + .newInstance(props); } catch (InstantiationException | IllegalAccessException | ClassNotFoundException ex) { - LOG.warn( - "Unable to invoke constructor with {} argument. " - + "ClassName = {}, interfaceName = {}, cause = {}, errorMessage = {}.", - constructorArgClass.getName(), - className, - iface.getName(), - ex.getCause(), - ex.getMessage()); + LOG.warn("Unable to invoke constructor with {} argument. ClassName = {}, interfaceName = {}, cause = {}, errorMessage = {}.", + constructorArgClass.getName(), className, intrface.getName(), ex.getCause(), ex.getMessage()); return null; } catch (NoSuchMethodException | InvocationTargetException e) { try { - LOG.info( - "Invoking default constructor. ClassName = {}, interfaceName = {}.", - className, - iface.getName()); + LOG.info("Invoking default constructor. ClassName = {}, interfaceName = {}.", className, intrface.getName()); // fallback to default constructor - instance = this.getClass().getClassLoader().loadClass(className).asSubclass(iface).newInstance(); + instance = this.getClass().getClassLoader().loadClass(className).asSubclass(intrface).newInstance(); } catch (InstantiationException | IllegalAccessException | ClassNotFoundException ex) { - LOG.error( - "Unable to invoke default constructor. " - + "ClassName = {}, interfaceName = {}, cause = {}, errorMessage = {}.", - className, - iface.getName(), - ex.getCause(), - ex.getMessage()); + LOG.error("Unable to invoke default constructor. ClassName = {}, interfaceName = {}, cause = {}, errorMessage = {}.", + className, intrface.getName(), ex.getCause(), ex.getMessage()); return null; } } } catch (ClassNotFoundException ex) { - LOG.error("The class does not exist. ClassName = {}, interfaceName = {}.", className, iface.getName()); + LOG.error("The class does not exist. ClassName = {}, interfaceName = {}.", className, intrface.getName()); return null; } catch (SecurityException ex) { - LOG.error( - "Unable to load class due to a security violation. " - + "ClassName = {}, interfaceName = {}, cause = {}, errorMessage = {}.", - className, - iface.getName(), - ex.getCause(), - ex.getMessage()); + LOG.error("Unable to load class due to a security violation. ClassName = {}, interfaceName = {}, cause = {}, errorMessage = {}.", + className, intrface.getName(), ex.getCause(), ex.getMessage()); return null; } diff --git a/broker/src/main/java/io/moquette/spi/impl/Qos0PublishHandler.java b/broker/src/main/java/io/moquette/spi/impl/Qos0PublishHandler.java index 26c180d6..c8bb0411 100644 --- a/broker/src/main/java/io/moquette/spi/impl/Qos0PublishHandler.java +++ b/broker/src/main/java/io/moquette/spi/impl/Qos0PublishHandler.java @@ -47,20 +47,12 @@ class Qos0PublishHandler extends QosPublishHandler { final int messageID = msg.variableHeader().messageId(); if (LOG.isTraceEnabled()) { - LOG.trace( - "Sending publish message to subscribers. " - + "MqttClientId = {}, topic = {}, messageId = {}, payload = {}, subscriptionTree = {}.", - clientID, - topic, - messageID, - DebugUtils.payload2Str(toStoreMsg.getMessage()), - subscriptions.dumpTree()); + LOG.trace("Sending publish message to subscribers. ClientId={}, topic={}, messageId={}, payload={}, " + + "subscriptionTree={}", clientID, topic, messageID, DebugUtils.payload2Str(toStoreMsg.getMessage()), + subscriptions.dumpTree()); } else { - LOG.info( - "Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}.", - clientID, - topic, - messageID); + LOG.info("Sending publish message to subscribers. ClientId={}, topic={}, messageId={}", clientID, topic, + messageID); } List topicMatchingSubscriptions = subscriptions.matches(topic); diff --git a/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java b/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java index c0534af2..21d22d76 100644 --- a/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java +++ b/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java @@ -57,37 +57,26 @@ class Qos1PublishHandler extends QosPublishHandler { toStoreMsg.setClientID(clientID); if (LOG.isTraceEnabled()) { - LOG.trace( - "Sending publish message to subscribers. " - + "MqttClientId = {}, topic = {}, messageId = {}, payload = {}, subscriptionTree = {}.", - clientID, - topic, - messageID, - DebugUtils.payload2Str(toStoreMsg.getMessage()), - subscriptions.dumpTree()); + LOG.trace("Sending publish message to subscribers. ClientId={}, topic={}, messageId={}, payload={}, " + + "subscriptionTree={}", clientID, topic, messageID, DebugUtils.payload2Str(toStoreMsg.getMessage()), + subscriptions.dumpTree()); } else { - LOG.info( - "Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}.", - clientID, - topic, - messageID); + LOG.info("Sending publish message to subscribers. ClientId={}, topic={}, messageId={}", clientID, topic, + messageID); } List topicMatchingSubscriptions = subscriptions.matches(topic); this.publisher.publish2Subscribers(toStoreMsg, topicMatchingSubscriptions); - // send PUBACK - // TODO Don't send PUBREC for Hz publish notification, if (msg.isLocal()) { sendPubAck(clientID, messageID); - // } if (msg.fixedHeader().isRetain()) { if (!msg.payload().isReadable()) { m_messagesStore.cleanRetained(topic); } else { // before wasn't stored - MessageGUID guid = m_messagesStore.storePublishForFuture(toStoreMsg); - m_messagesStore.storeRetained(topic, guid); + //MessageGUID guid = m_messagesStore.storePublishForFuture(toStoreMsg); + m_messagesStore.storeRetained(topic, toStoreMsg.getGuid()); } } @@ -102,17 +91,13 @@ class Qos1PublishHandler extends QosPublishHandler { try { if (connectionDescriptors == null) { - throw new RuntimeException( - "Internal bad error, found connectionDescriptors to null while it should be initialized, " - + "somewhere it's overwritten!!"); + throw new RuntimeException("Internal bad error, found connectionDescriptors to null while it should " + + "be initialized, somewhere it's overwritten!!"); } LOG.debug("clientIDs are {}", connectionDescriptors); if (!connectionDescriptors.isConnected(clientId)) { - throw new RuntimeException( - String.format( - "Can't find a ConnectionDescriptor for client %s in cache %s", - clientId, - connectionDescriptors)); + throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client %s in cache %s", + clientId,connectionDescriptors)); } connectionDescriptors.sendMessage(pubAckMessage, messageID, clientId); } catch (Throwable t) { diff --git a/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java b/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java index b42d1b4e..063f66f2 100644 --- a/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java +++ b/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java @@ -58,27 +58,19 @@ class Qos2PublishHandler extends QosPublishHandler { toStoreMsg.setClientID(clientID); if (LOG.isTraceEnabled()) { - LOG.trace( - "Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}, " - + "payload = {}, subscriptionTree = {}.", - clientID, - topic, - messageID, - DebugUtils.payload2Str(toStoreMsg.getMessage()), - subscriptions.dumpTree()); + LOG.trace( "Sending publish message to subscribers. ClientId={}, topic={}, messageId={}, payload={}, " + + "subscriptionTree={}", clientID, topic, messageID, DebugUtils.payload2Str(toStoreMsg.getMessage()), + subscriptions.dumpTree()); } else { - LOG.info( - "Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}.", - clientID, - topic, - messageID); + LOG.info("Sending publish message to subscribers. ClientId={}, topic={}, messageId={}", clientID, topic, + messageID); } // QoS2 MessageGUID guid = m_messagesStore.storePublishForFuture(toStoreMsg); // TODO Don't send PUBREC for Hz publish notification, if (msg.isLocal()) { sendPubRec(clientID, messageID); - // } + // Next the client will send us a pub rel // NB publish to subscribers for QoS 2 happen upon PUBREL from publisher @@ -100,27 +92,19 @@ class Qos2PublishHandler extends QosPublishHandler { void processPubRel(Channel channel, MqttMessage msg) { String clientID = NettyUtils.clientID(channel); int messageID = messageId(msg); - LOG.info("Processing PUBREL message. MqttClientId = {}, messageId = {}.", clientID, messageID); + LOG.info("Processing PUBREL message. ClientId={}, messageId={}", clientID, messageID); ClientSession targetSession = m_sessionsStore.sessionForClient(clientID); IMessagesStore.StoredMessage evt = targetSession.storedMessage(messageID); final Topic topic = new Topic(evt.getTopic()); List topicMatchingSubscriptions = subscriptions.matches(topic); if (LOG.isDebugEnabled()) { - LOG.debug( - "Sending publish message to subscribers. " - + "MqttClientId = {}, topic = {}, messageId = {}, payload = {}, subscriptionTree = {}.", - clientID, - topic, - messageID, - DebugUtils.payload2Str(evt.getMessage()), - subscriptions.dumpTree()); + LOG.debug("Sending publish message to subscribers. ClientId={}, topic={}, messageId={}, payload={}, " + + "subscriptionTree={}", clientID, topic, messageID, DebugUtils.payload2Str(evt.getMessage()), + subscriptions.dumpTree()); } else { - LOG.info( - "Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}.", - clientID, - topic, - messageID); + LOG.info("Sending publish message to subscribers. ClientId={}, topic={}, messageId={}", clientID, topic, + messageID); } this.publisher.publish2Subscribers(evt, topicMatchingSubscriptions); @@ -136,14 +120,14 @@ class Qos2PublishHandler extends QosPublishHandler { } private void sendPubRec(String clientID, int messageID) { - LOG.debug("Sending PUBREC message. MqttClientId = {}, messageId = {}.", clientID, messageID); + LOG.debug("Sending PUBREC message. ClientId={}, messageId={}", clientID, messageID); MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBREC, false, AT_MOST_ONCE, false, 0); MqttMessage pubRecMessage = new MqttMessage(fixedHeader, from(messageID)); connectionDescriptors.sendMessage(pubRecMessage, messageID, clientID); } private void sendPubComp(String clientID, int messageID) { - LOG.debug("Sending PUBCOMP message. MqttClientId = {}, messageId = {}.", clientID, messageID); + LOG.debug("Sending PUBCOMP message. ClientId={}, messageId={}", clientID, messageID); MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBCOMP, false, AT_MOST_ONCE, false, 0); MqttMessage pubCompMessage = new MqttMessage(fixedHeader, from(messageID)); connectionDescriptors.sendMessage(pubCompMessage, messageID, clientID); diff --git a/broker/src/main/java/io/moquette/spi/impl/subscriptions/Topic.java b/broker/src/main/java/io/moquette/spi/impl/subscriptions/Topic.java index 753af328..8a74d138 100644 --- a/broker/src/main/java/io/moquette/spi/impl/subscriptions/Topic.java +++ b/broker/src/main/java/io/moquette/spi/impl/subscriptions/Topic.java @@ -47,7 +47,7 @@ public class Topic implements Serializable { valid = true; } catch (ParseException e) { valid = false; - LOG.error(e.toString(), e); + LOG.error("Error parsing the topic: {}, message: {}", topic, e.getMessage()); } } diff --git a/broker/src/test/java/io/moquette/server/ConfigurationClassLoaderTest.java b/broker/src/test/java/io/moquette/server/ConfigurationClassLoaderTest.java index 1b05c885..41dda657 100644 --- a/broker/src/test/java/io/moquette/server/ConfigurationClassLoaderTest.java +++ b/broker/src/test/java/io/moquette/server/ConfigurationClassLoaderTest.java @@ -20,9 +20,9 @@ import io.moquette.BrokerConstants; import io.moquette.server.config.IConfig; import io.moquette.server.config.MemoryConfig; import io.moquette.spi.impl.subscriptions.Topic; -import io.moquette.spi.impl.MemoryMessagesStore; -import io.moquette.spi.impl.MemoryStorageService; -import io.moquette.spi.persistence.MemorySessionStore; +import io.moquette.persistence.MemoryMessagesStore; +import io.moquette.persistence.MemoryStorageService; +import io.moquette.persistence.MemorySessionStore; import io.moquette.spi.security.IAuthenticator; import io.moquette.spi.security.IAuthorizator; import org.junit.After; diff --git a/broker/src/test/java/io/moquette/server/IntegrationUtils.java b/broker/src/test/java/io/moquette/server/IntegrationUtils.java index cac6c956..9349e0b2 100644 --- a/broker/src/test/java/io/moquette/server/IntegrationUtils.java +++ b/broker/src/test/java/io/moquette/server/IntegrationUtils.java @@ -55,7 +55,7 @@ public class IntegrationUtils { return testProperties; } - public static void cleanPersistenceFile(IConfig config) { + /*public static void cleanPersistenceFile(IConfig config) { String fileName = config.getProperty(PERSISTENT_STORE_PROPERTY_NAME); cleanPersistenceFile(fileName); } @@ -68,5 +68,5 @@ public class IntegrationUtils { new File(fileName + ".t").delete(); } assertFalse(dbFile.exists()); - } + }*/ } diff --git a/broker/src/test/java/io/moquette/server/ServerIntegrationDBAuthenticatorTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationDBAuthenticatorTest.java index 1e23b545..81c7cb83 100644 --- a/broker/src/test/java/io/moquette/server/ServerIntegrationDBAuthenticatorTest.java +++ b/broker/src/test/java/io/moquette/server/ServerIntegrationDBAuthenticatorTest.java @@ -53,7 +53,6 @@ public class ServerIntegrationDBAuthenticatorTest { private void stopServer() { m_server.stopServer(); - IntegrationUtils.cleanPersistenceFile(m_config); } private Properties addDBAuthenticatorConf(Properties properties) { @@ -67,9 +66,6 @@ public class ServerIntegrationDBAuthenticatorTest { @Before public void setUp() throws Exception { - String dbPath = IntegrationUtils.localMapDBPath(); - IntegrationUtils.cleanPersistenceFile(dbPath); - startServer(); m_client = new MqttClient("tcp://localhost:1883", "TestClient", s_dataStore); diff --git a/broker/src/test/java/io/moquette/server/ServerIntegrationEmbeddedPublishTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationEmbeddedPublishTest.java index 5689d9c5..209b7b7a 100644 --- a/broker/src/test/java/io/moquette/server/ServerIntegrationEmbeddedPublishTest.java +++ b/broker/src/test/java/io/moquette/server/ServerIntegrationEmbeddedPublishTest.java @@ -62,9 +62,6 @@ public class ServerIntegrationEmbeddedPublishTest { @Before public void setUp() throws Exception { - String dbPath = IntegrationUtils.localMapDBPath(); - IntegrationUtils.cleanPersistenceFile(dbPath); - startServer(); m_subscriber = new MqttClient("tcp://localhost:1883", "TestClient", s_dataStore); @@ -79,7 +76,6 @@ public class ServerIntegrationEmbeddedPublishTest { } m_server.stopServer(); - IntegrationUtils.cleanPersistenceFile(m_config); } private void subscribeToWithQos(String topic, int qos) throws Exception { diff --git a/broker/src/test/java/io/moquette/server/ServerIntegrationFuseTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationFuseTest.java index 3bf83814..3a4f4e36 100644 --- a/broker/src/test/java/io/moquette/server/ServerIntegrationFuseTest.java +++ b/broker/src/test/java/io/moquette/server/ServerIntegrationFuseTest.java @@ -69,7 +69,6 @@ public class ServerIntegrationFuseTest { } m_server.stopServer(); - IntegrationUtils.cleanPersistenceFile(m_config); } @Test diff --git a/broker/src/test/java/io/moquette/server/ServerIntegrationHazelcastHandlerInterceptorTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationHazelcastHandlerInterceptorTest.java index eb60737c..9f8efbb7 100644 --- a/broker/src/test/java/io/moquette/server/ServerIntegrationHazelcastHandlerInterceptorTest.java +++ b/broker/src/test/java/io/moquette/server/ServerIntegrationHazelcastHandlerInterceptorTest.java @@ -67,9 +67,6 @@ public class ServerIntegrationHazelcastHandlerInterceptorTest { @Before public void setUp() throws Exception { - String dbPath = IntegrationUtils.localMapDBPath(); - IntegrationUtils.cleanPersistenceFile(dbPath); - final Properties configProps = addHazelCastConf( IntegrationUtils.prepareTestClusterProperties(1883), 1883, diff --git a/broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java index 39d61ec7..8cd39f07 100644 --- a/broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java +++ b/broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java @@ -57,9 +57,6 @@ public class ServerIntegrationPahoTest { @Before public void setUp() throws Exception { - String dbPath = IntegrationUtils.localMapDBPath(); - IntegrationUtils.cleanPersistenceFile(dbPath); - startServer(); m_client = new MqttClient("tcp://localhost:1883", "TestClient", s_dataStore); @@ -84,7 +81,6 @@ public class ServerIntegrationPahoTest { private void stopServer() { m_server.stopServer(); - IntegrationUtils.cleanPersistenceFile(m_config); } @Test @@ -176,6 +172,7 @@ public class ServerIntegrationPahoTest { m_client.disconnect(); } + @Ignore("This test hasn't any meaning using in memory storage service") @Test public void testCleanSession_maintainClientSubscriptions_withServerRestart() throws Exception { LOG.info("*** testCleanSession_maintainClientSubscriptions_withServerRestart ***"); diff --git a/broker/src/test/java/io/moquette/server/ServerIntegrationQoSValidationTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationQoSValidationTest.java index 0e83a65e..7643460c 100644 --- a/broker/src/test/java/io/moquette/server/ServerIntegrationQoSValidationTest.java +++ b/broker/src/test/java/io/moquette/server/ServerIntegrationQoSValidationTest.java @@ -89,7 +89,6 @@ public class ServerIntegrationQoSValidationTest { } m_server.stopServer(); - IntegrationUtils.cleanPersistenceFile(m_config); } @Test diff --git a/broker/src/test/java/io/moquette/server/ServerIntegrationWebSocketTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationWebSocketTest.java index 2946234c..4d073ead 100644 --- a/broker/src/test/java/io/moquette/server/ServerIntegrationWebSocketTest.java +++ b/broker/src/test/java/io/moquette/server/ServerIntegrationWebSocketTest.java @@ -65,7 +65,6 @@ public class ServerIntegrationWebSocketTest { client.stop(); m_server.stopServer(); - IntegrationUtils.cleanPersistenceFile(m_config); } @Test diff --git a/broker/src/test/java/io/moquette/server/ServerLowlevelMessagesIntegrationTests.java b/broker/src/test/java/io/moquette/server/ServerLowlevelMessagesIntegrationTests.java index 3a0431df..b095cac7 100644 --- a/broker/src/test/java/io/moquette/server/ServerLowlevelMessagesIntegrationTests.java +++ b/broker/src/test/java/io/moquette/server/ServerLowlevelMessagesIntegrationTests.java @@ -70,7 +70,6 @@ public class ServerLowlevelMessagesIntegrationTests { Thread.sleep(300); // to let the close event pass before server stop event m_server.stopServer(); LOG.debug("After asked server to stop"); - IntegrationUtils.cleanPersistenceFile(m_config); } @Test @@ -93,22 +92,11 @@ public class ServerLowlevelMessagesIntegrationTests { } private static MqttConnectMessage createConnectMessage(String clientID, int keepAlive) { - MqttFixedHeader mqttFixedHeader = new MqttFixedHeader( - MqttMessageType.CONNECT, - false, - MqttQoS.AT_MOST_ONCE, - false, - 0); + MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.CONNECT, false, MqttQoS.AT_MOST_ONCE, + false, 0); MqttConnectVariableHeader mqttConnectVariableHeader = new MqttConnectVariableHeader( - MqttVersion.MQTT_3_1.protocolName(), - MqttVersion.MQTT_3_1.protocolLevel(), - false, - false, - false, - 1, - false, - true, - keepAlive); + MqttVersion.MQTT_3_1.protocolName(), MqttVersion.MQTT_3_1.protocolLevel(), false, false, false, 1, false, + true, keepAlive); MqttConnectPayload mqttConnectPayload = new MqttConnectPayload(clientID, null, null, null, null); return new MqttConnectMessage(mqttFixedHeader, mqttConnectVariableHeader, mqttConnectPayload); } diff --git a/broker/src/test/java/io/moquette/server/ServerRestartIntegrationTest.java b/broker/src/test/java/io/moquette/server/ServerRestartIntegrationTest.java index f8bbc583..65614015 100644 --- a/broker/src/test/java/io/moquette/server/ServerRestartIntegrationTest.java +++ b/broker/src/test/java/io/moquette/server/ServerRestartIntegrationTest.java @@ -64,7 +64,6 @@ public class ServerRestartIntegrationTest { @Before public void setUp() throws Exception { String dbPath = IntegrationUtils.localMapDBPath(); - IntegrationUtils.cleanPersistenceFile(dbPath); startServer(); m_subscriber = new MqttClient("tcp://localhost:1883", "Subscriber", s_dataStore); @@ -85,7 +84,6 @@ public class ServerRestartIntegrationTest { } m_server.stopServer(); - IntegrationUtils.cleanPersistenceFile(m_config); } @Test @@ -97,7 +95,6 @@ public class ServerRestartIntegrationTest { // shutdown the server m_server.stopServer(); - IntegrationUtils.cleanPersistenceFile(m_config); // restart the server m_server.startServer(IntegrationUtils.prepareTestProperties()); diff --git a/broker/src/test/java/io/moquette/spi/ClientSessionTest.java b/broker/src/test/java/io/moquette/spi/ClientSessionTest.java index 2f2505f3..69d1e568 100644 --- a/broker/src/test/java/io/moquette/spi/ClientSessionTest.java +++ b/broker/src/test/java/io/moquette/spi/ClientSessionTest.java @@ -1,7 +1,7 @@ package io.moquette.spi; -import io.moquette.spi.impl.MemoryStorageService; +import io.moquette.persistence.MemoryStorageService; import io.moquette.spi.impl.subscriptions.Subscription; import io.moquette.spi.impl.subscriptions.SubscriptionsStore; import io.moquette.spi.impl.subscriptions.Topic; diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java index 925d0815..50e42f37 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java @@ -17,6 +17,7 @@ package io.moquette.spi.impl; import io.moquette.interception.InterceptHandler; +import io.moquette.persistence.MemoryStorageService; import io.moquette.server.netty.MessageBuilder; import io.moquette.server.netty.NettyUtils; import io.moquette.spi.IMatchingCondition; @@ -98,14 +99,8 @@ public class ProtocolProcessorTest { subscriptions = new SubscriptionsStore(); subscriptions.init(memStorage.sessionsStore()); m_processor = new ProtocolProcessor(); - m_processor.init( - subscriptions, - m_messagesStore, - m_sessionStore, - m_mockAuthenticator, - true, - new PermitAllAuthorizator(), - NO_OBSERVERS_INTERCEPTOR); + m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, true, + new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); } @Test @@ -176,13 +171,7 @@ public class ProtocolProcessorTest { // simulate a connect that register a clientID to an IoSession MemoryStorageService storageService = new MemoryStorageService(); subs.init(storageService.sessionsStore()); - m_processor.init( - subs, - m_messagesStore, - m_sessionStore, - null, - true, - new PermitAllAuthorizator(), + m_processor.init(subs, m_messagesStore, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); EmbeddedChannel firstReceiverChannel = new EmbeddedChannel(); @@ -228,10 +217,8 @@ public class ProtocolProcessorTest { // Verify assertTrue(m_channel.readOutbound() instanceof MqttSubAckMessage); - Subscription expectedSubscription = new Subscription( - FAKE_CLIENT_ID, - new Topic(FAKE_TOPIC), - MqttQoS.AT_MOST_ONCE); + Subscription expectedSubscription = new Subscription(FAKE_CLIENT_ID, new Topic(FAKE_TOPIC), + MqttQoS.AT_MOST_ONCE); assertTrue(subscriptions.contains(expectedSubscription)); } @@ -244,13 +231,7 @@ public class ProtocolProcessorTest { when(mockAuthorizator.canRead(eq(new Topic(FAKE_TOPIC)), eq(fakeUserName), eq(FAKE_CLIENT_ID))) .thenReturn(false); - m_processor.init( - subscriptions, - m_messagesStore, - m_sessionStore, - m_mockAuthenticator, - true, - mockAuthorizator, + m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, true, mockAuthorizator, NO_OBSERVERS_INTERCEPTOR); // Exercise @@ -288,10 +269,8 @@ public class ProtocolProcessorTest { // Verify assertEquals(1, subscriptions.size()); - Subscription expectedSubscription = new Subscription( - FAKE_CLIENT_ID, - new Topic(FAKE_TOPIC), - MqttQoS.AT_MOST_ONCE); + Subscription expectedSubscription = new Subscription(FAKE_CLIENT_ID, new Topic(FAKE_TOPIC), + MqttQoS.AT_MOST_ONCE); assertTrue(subscriptions.contains(expectedSubscription)); } @@ -332,10 +311,7 @@ public class ProtocolProcessorTest { @Test public void testPublishOfRetainedMessage_afterNewSubscription() throws Exception { // simulate a connect that register a clientID to an IoSession - final Subscription subscription = new Subscription( - FAKE_PUBLISHER_ID, - new Topic(FAKE_TOPIC), - MqttQoS.AT_MOST_ONCE); + final Subscription subscription = new Subscription(FAKE_PUBLISHER_ID, new Topic(FAKE_TOPIC), MqttQoS.AT_MOST_ONCE); // subscriptions.matches(topic) redefine the method to return true SubscriptionsStore subs = new SubscriptionsStore() { @@ -353,13 +329,7 @@ public class ProtocolProcessorTest { subs.init(storageService.sessionsStore()); // simulate a connect that register a clientID to an IoSession - m_processor.init( - subs, - m_messagesStore, - m_sessionStore, - null, - true, - new PermitAllAuthorizator(), + m_processor.init(subs, m_messagesStore, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); MqttConnectMessage connectMessage = MessageBuilder.connect().clientId(FAKE_PUBLISHER_ID) .protocolVersion(MqttVersion.MQTT_3_1).cleanSession(true).build(); @@ -398,13 +368,7 @@ public class ProtocolProcessorTest { retainedMessage.setClientID(FAKE_PUBLISHER_ID); m_messagesStore.storePublishForFuture(retainedMessage); - m_processor.init( - subs, - m_messagesStore, - m_sessionStore, - null, - true, - new PermitAllAuthorizator(), + m_processor.init(subs, m_messagesStore, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); MqttConnectMessage connectMessage = MessageBuilder.connect().clientId(FAKE_PUBLISHER_ID) @@ -427,13 +391,7 @@ public class ProtocolProcessorTest { List inactiveSubscriptions = Collections.singletonList(inactiveSub); when(mockedSubscriptions.matches(eq(new Topic("/topic")))).thenReturn(inactiveSubscriptions); m_processor = new ProtocolProcessor(); - m_processor.init( - mockedSubscriptions, - m_messagesStore, - m_sessionStore, - null, - true, - new PermitAllAuthorizator(), + m_processor.init(mockedSubscriptions, m_messagesStore, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); // Exercise diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java index ce6bdeb1..b13be265 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java @@ -16,6 +16,7 @@ package io.moquette.spi.impl; +import io.moquette.persistence.MemoryStorageService; import io.moquette.server.netty.MessageBuilder; import io.moquette.server.netty.NettyUtils; import io.moquette.spi.ClientSession; diff --git a/broker/src/test/java/io/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java b/broker/src/test/java/io/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java index 4617cbc9..e7651f03 100644 --- a/broker/src/test/java/io/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java @@ -19,7 +19,7 @@ package io.moquette.spi.impl.subscriptions; import io.moquette.spi.ClientSession; import io.moquette.spi.ISessionsStore; import io.moquette.spi.ISessionsStore.ClientTopicCouple; -import io.moquette.spi.impl.MemoryStorageService; +import io.moquette.persistence.MemoryStorageService; import io.netty.handler.codec.mqtt.MqttQoS; import org.junit.Before; import org.junit.Test; diff --git a/mapdb_storage/pom.xml b/mapdb_storage/pom.xml new file mode 100644 index 00000000..f276cffc --- /dev/null +++ b/mapdb_storage/pom.xml @@ -0,0 +1,164 @@ + + 4.0.0 + + ../ + moquette-parent + io.moquette + 0.10-SNAPSHOT + + + moquette-mapdb-storage + jar + Moquette - MapDB storage + + + UTF-8 + io.moquette.moquette-broker + + + + + ${project.groupId} + moquette-broker + ${project.version} + + + + org.slf4j + slf4j-api + 1.7.5 + + + + org.mapdb + mapdb + + + + 1.0.8 + + + + org.hdrhistogram + HdrHistogram + 2.1.4 + + + + com.hazelcast + hazelcast + 3.5.4 + + + + + junit + junit + 4.8.2 + test + + + + org.slf4j + slf4j-log4j12 + 1.7.5 + test + + + + + org.mockito + mockito-core + 1.9.0 + test + + + + + + + + + + + + + + + + + + + + + + + + + org.assertj + assertj-core + + 2.6.0 + test + + + + + + + + + + + + + + org.apache.maven.plugins + maven-surefire-plugin + + + **/*Test.java + + + + + + + maven-jar-plugin + + + ${project.build.outputDirectory}/META-INF/MANIFEST.MF + + + + + + test-jar + + + + + + org.apache.felix + maven-bundle-plugin + + + bundle-manifest + process-classes + + manifest + + + + + + + + + + bintray + https://api.bintray.com/maven/${bintray.repo}/${bintray.package} + + + diff --git a/broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java b/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBMessagesStore.java similarity index 99% rename from broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java rename to mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBMessagesStore.java index 79dd0389..dad74093 100644 --- a/broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java +++ b/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBMessagesStore.java @@ -14,7 +14,7 @@ * You may elect to redistribute this code under either of these licenses. */ -package io.moquette.spi.persistence; +package io.moquette.persistence.mapdb; import io.moquette.spi.IMatchingCondition; import io.moquette.spi.IMessagesStore; diff --git a/broker/src/main/java/io/moquette/spi/persistence/MapDBPersistentStore.java b/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBPersistentStore.java similarity index 85% rename from broker/src/main/java/io/moquette/spi/persistence/MapDBPersistentStore.java rename to mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBPersistentStore.java index 7d01ce95..6e42657d 100644 --- a/broker/src/main/java/io/moquette/spi/persistence/MapDBPersistentStore.java +++ b/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBPersistentStore.java @@ -14,8 +14,9 @@ * You may elect to redistribute this code under either of these licenses. */ -package io.moquette.spi.persistence; +package io.moquette.persistence.mapdb; +import io.moquette.BrokerConstants; import io.moquette.server.config.IConfig; import io.moquette.spi.IMessagesStore; import io.moquette.spi.ISessionsStore; @@ -29,28 +30,12 @@ import java.io.IOException; import java.io.Serializable; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; -import static io.moquette.BrokerConstants.AUTOSAVE_INTERVAL_PROPERTY_NAME; -import static io.moquette.BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME; /** * MapDB main persistence implementation */ public class MapDBPersistentStore implements IStore { - /** - * This is a DTO used to persist minimal status (clean session and activation status) of a - * session. - */ - public static class PersistentSession implements Serializable { - - private static final long serialVersionUID = 5052054783220481854L; - public final boolean cleanSession; - - public PersistentSession(boolean cleanSession) { - this.cleanSession = cleanSession; - } - } - private static final Logger LOG = LoggerFactory.getLogger(MapDBPersistentStore.class); private DB m_db; @@ -62,8 +47,8 @@ public class MapDBPersistentStore implements IStore { private ISessionsStore m_sessionsStore; public MapDBPersistentStore(IConfig props, ScheduledExecutorService scheduler) { - this.m_storePath = props.getProperty(PERSISTENT_STORE_PROPERTY_NAME, ""); - this.m_autosaveInterval = Integer.parseInt(props.getProperty(AUTOSAVE_INTERVAL_PROPERTY_NAME, "30")); + this.m_storePath = props.getProperty(BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME, ""); + this.m_autosaveInterval = Integer.parseInt(props.getProperty(BrokerConstants.AUTOSAVE_INTERVAL_PROPERTY_NAME, "30")); this.m_scheduler = scheduler; } @@ -135,6 +120,9 @@ public class MapDBPersistentStore implements IStore { LOG.info("Closing MapDB store..."); this.m_db.close(); LOG.info("Stopping MapDB commit tasks..."); + + //TODO th scheduler must be stopped by the owning (the instance of Server) + //invalidate the added task this.m_scheduler.shutdown(); try { m_scheduler.awaitTermination(10L, TimeUnit.SECONDS); diff --git a/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java b/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBSessionsStore.java similarity index 77% rename from broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java rename to mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBSessionsStore.java index 2f3b9c8a..70fe89b8 100644 --- a/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java +++ b/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBSessionsStore.java @@ -14,8 +14,9 @@ * You may elect to redistribute this code under either of these licenses. */ -package io.moquette.spi.persistence; +package io.moquette.persistence.mapdb; +import io.moquette.persistence.PersistentSession; import io.moquette.spi.ClientSession; import io.moquette.spi.IMessagesStore; import io.moquette.spi.IMessagesStore.StoredMessage; @@ -24,7 +25,6 @@ import io.moquette.spi.MessageGUID; import io.moquette.spi.impl.Utils; import io.moquette.spi.impl.subscriptions.Subscription; import io.moquette.spi.impl.subscriptions.Topic; -import io.moquette.spi.persistence.MapDBPersistentStore.PersistentSession; import org.mapdb.DB; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -68,47 +68,38 @@ class MapDBSessionsStore implements ISessionsStore { @Override public void addNewSubscription(Subscription newSubscription) { - LOG.info( - "Adding new subscription. CId= {}, topics = {}.", - newSubscription.getClientId(), - newSubscription.getTopicFilter()); + LOG.info("Adding new subscription. ClientId={}, topics={}", newSubscription.getClientId(), + newSubscription.getTopicFilter()); final String clientID = newSubscription.getClientId(); m_db.getHashMap("subscriptions_" + clientID).put(newSubscription.getTopicFilter(), newSubscription); if (LOG.isTraceEnabled()) { - LOG.trace( - "The subscription has been added. CId= {}, topics = {}, clientSubscriptions = {}.", - newSubscription.getClientId(), - newSubscription.getTopicFilter(), - m_db.getHashMap("subscriptions_" + clientID)); + LOG.trace("Subscription has been added. ClientId={}, topics={}, clientSubscriptions={}", + newSubscription.getClientId(), newSubscription.getTopicFilter(), + m_db.getHashMap("subscriptions_" + clientID)); } } @Override public void removeSubscription(Topic topicFilter, String clientID) { - LOG.info("Removing subscription. CId= {}, topics = {}.", clientID, topicFilter); + LOG.info("Removing subscription. ClientId={}, topics={}", clientID, topicFilter); if (!m_db.exists("subscriptions_" + clientID)) { return; } m_db.getHashMap("subscriptions_" + clientID).remove(topicFilter); if (LOG.isDebugEnabled()) { - LOG.debug( - "The subscription has been removed. CId= {}, topics = {}, clientSubscriptions = {}.", - clientID, - topicFilter, - m_db.getHashMap("subscriptions_" + clientID)); + LOG.debug("Subscription has been removed. ClientId={}, topics={}, clientSubscriptions={}", clientID, + topicFilter, m_db.getHashMap("subscriptions_" + clientID)); } } @Override public void wipeSubscriptions(String clientID) { - LOG.info("Wiping subscriptions. CId= {}.", clientID); + LOG.info("Wiping subscriptions. ClientId={}", clientID); m_db.delete("subscriptions_" + clientID); if (LOG.isDebugEnabled()) { - LOG.debug( - "The subscriptions have been removed. CId= {}, clientSubscriptions = {}.", - clientID, - m_db.getHashMap("subscriptions_" + clientID)); + LOG.debug("Subscriptions have been removed. ClientId={}, clientSubscriptions={}", clientID, + m_db.getHashMap("subscriptions_" + clientID)); } } @@ -123,7 +114,7 @@ class MapDBSessionsStore implements ISessionsStore { } } 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; } @@ -131,7 +122,7 @@ class MapDBSessionsStore implements ISessionsStore { @Override public Subscription getSubscription(ClientTopicCouple couple) { ConcurrentMap clientSubscriptions = m_db.getHashMap("subscriptions_" + couple.clientID); - LOG.info("Retrieving subscriptions. CId= {}, subscriptions = {}.", couple.clientID, clientSubscriptions); + LOG.info("Retrieving subscriptions. ClientId={}, subscriptions={}", couple.clientID, clientSubscriptions); return clientSubscriptions.get(couple.topicFilter); } @@ -143,7 +134,7 @@ class MapDBSessionsStore implements ISessionsStore { ConcurrentMap clientSubscriptions = m_db.getHashMap("subscriptions_" + clientID); 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; } @@ -155,22 +146,20 @@ class MapDBSessionsStore implements ISessionsStore { @Override public ClientSession createNewSession(String clientID, boolean cleanSession) { if (m_persistentSessions.containsKey(clientID)) { - LOG.error( - "Unable to create a new session: the client ID is already in use. CId= {}, cleanSession = {}.", - clientID, - cleanSession); + LOG.error("Unable to create a new session: the client ID is already in use. ClientId={}, cleanSession={}", + clientID, cleanSession); throw new IllegalArgumentException("Can't create a session with the ID of an already existing" + clientID); } - LOG.info("Creating new session. CId= {}, cleanSession = {}.", clientID, cleanSession); + LOG.info("Creating new session. ClientId={}, cleanSession={}", clientID, cleanSession); m_persistentSessions.putIfAbsent(clientID, new PersistentSession(cleanSession)); return new ClientSession(clientID, m_messagesStore, this, cleanSession); } @Override public ClientSession sessionForClient(String clientID) { - LOG.info("Retrieving session. CId= {}.", clientID); + LOG.info("Retrieving session. ClientId={}", clientID); if (!m_persistentSessions.containsKey(clientID)) { - LOG.warn("The session does not exist. CId= {}.", clientID); + LOG.warn("The session does not exist. ClientId={}", clientID); return null; } @@ -189,8 +178,8 @@ class MapDBSessionsStore implements ISessionsStore { @Override public void updateCleanStatus(String clientID, boolean cleanSession) { - LOG.info("Updating cleanSession flag. CId= {}, cleanSession = {}.", clientID, cleanSession); - m_persistentSessions.put(clientID, new MapDBPersistentStore.PersistentSession(cleanSession)); + LOG.info("Updating cleanSession flag. ClientId={}, cleanSession={}", clientID, cleanSession); + m_persistentSessions.put(clientID, new PersistentSession(cleanSession)); } /** @@ -199,7 +188,7 @@ class MapDBSessionsStore implements ISessionsStore { @Override public int nextPacketID(String clientID) { if (LOG.isDebugEnabled()) { - LOG.debug("Generating next packet ID. CId= {}.", clientID); + LOG.debug("Generating next packet ID. ClientId={}", clientID); } Set inFlightForClient = this.m_inFlightIds.get(clientID); if (inFlightForClient == null) { @@ -215,17 +204,17 @@ class MapDBSessionsStore implements ISessionsStore { int nextPacketId = (maxId % 0xFFFF) + 1; inFlightForClient.add(nextPacketId); 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. ClientId={}, result={}", clientID, nextPacketId); } return nextPacketId; } @Override public void inFlightAck(String clientID, int messageID) { - LOG.debug("Acknowledging inflight message. CId= {}, messageId = {}.", clientID, messageID); + LOG.debug("Acknowledging inflight message. ClientId={}, messageId={}", clientID, messageID); Map m = this.m_inflightStore.get(clientID); if (m == null) { - LOG.warn("Unable to retrieve inflight message record. CId= {}, messageId = {}.", clientID, messageID); + LOG.warn("Unable to retrieve inflight message record. ClientId={}, messageId={}", clientID, messageID); return; } m.remove(messageID); @@ -239,7 +228,7 @@ class MapDBSessionsStore implements ISessionsStore { @Override 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. ClientId={}, messageId={}, guid={}", clientID, messageID, guid); ConcurrentMap m = this.m_inflightStore.get(clientID); if (m == null) { m = new ConcurrentHashMap<>(); @@ -251,24 +240,24 @@ class MapDBSessionsStore implements ISessionsStore { @Override public BlockingQueue queue(String clientID) { - LOG.info("Queuing pending message. CId= {}, guid = {}.", clientID); + LOG.info("Queuing pending message. ClientId={}, guid={}", clientID); return this.m_db.getQueue(clientID); } @Override public void dropQueue(String clientID) { - LOG.info("Removing pending messages. CId= {}.", clientID); + LOG.info("Removing pending messages. ClientId={}", clientID); this.m_db.delete(clientID); } @Override public void moveInFlightToSecondPhaseAckWaiting(String clientID, int messageID) { if (LOG.isDebugEnabled()) { - LOG.debug("Moving inflight message to 2nd phase ack state. CId= {}, messageID = {}.", clientID, messageID); + LOG.debug("Moving inflight message to 2nd phase ack state. ClientId={}, messageID={}", clientID, messageID); } Map m = this.m_inflightStore.get(clientID); if (m == null) { - LOG.warn("Unable to retrieve inflight message record. CId= {}, messageId = {}.", clientID, messageID); + LOG.warn("Unable to retrieve inflight message record. ClientId={}, messageId={}", clientID, messageID); return; } MessageGUID guid = m.remove(messageID); @@ -279,8 +268,8 @@ class MapDBSessionsStore implements ISessionsStore { inFlightForClient.remove(messageID); } - Map messageIDs = Utils - .defaultGet(m_secondPhaseStore, clientID, new HashMap()); + final HashMap emptyGuids = new HashMap<>(); + Map messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, emptyGuids); messageIDs.put(messageID, guid); m_secondPhaseStore.put(clientID, messageIDs); } @@ -290,8 +279,8 @@ class MapDBSessionsStore implements ISessionsStore { if (LOG.isDebugEnabled()) { LOG.debug("Processing second phase ACK CId={}, messageId={}", clientID, messageID); } - Map messageIDs = Utils - .defaultGet(m_secondPhaseStore, clientID, new HashMap()); + final HashMap emptyGuids = new HashMap<>(); + Map messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, emptyGuids); MessageGUID guid = messageIDs.remove(messageID); m_secondPhaseStore.put(clientID, messageIDs); return guid; @@ -307,7 +296,7 @@ class MapDBSessionsStore implements ISessionsStore { } MessageGUID guid = clientEntries.get(messageID); 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. ClientId={}, messageId={}", clientID, messageID); return null; } return m_messagesStore.getMessageByGuid(guid); diff --git a/broker/src/test/java/io/moquette/spi/persistence/MapDBPersistentStoreTest.java b/mapdb_storage/src/test/java/io/moquette/persistence/mapdb/MapDBPersistentStoreTest.java similarity index 91% rename from broker/src/test/java/io/moquette/spi/persistence/MapDBPersistentStoreTest.java rename to mapdb_storage/src/test/java/io/moquette/persistence/mapdb/MapDBPersistentStoreTest.java index decfd51e..ab83c8a7 100644 --- a/broker/src/test/java/io/moquette/spi/persistence/MapDBPersistentStoreTest.java +++ b/mapdb_storage/src/test/java/io/moquette/persistence/mapdb/MapDBPersistentStoreTest.java @@ -14,12 +14,12 @@ * You may elect to redistribute this code under either of these licenses. */ -package io.moquette.spi.persistence; +package io.moquette.persistence.mapdb; +import java.io.File; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import io.moquette.BrokerConstants; -import io.moquette.server.IntegrationUtils; import io.moquette.server.config.IConfig; import io.moquette.server.config.MemoryConfig; import io.moquette.spi.ClientSession; @@ -36,6 +36,8 @@ import org.junit.Test; import java.util.List; import java.util.Properties; import java.util.concurrent.TimeUnit; + +import static io.moquette.BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME; import static org.junit.Assert.*; /** @@ -55,7 +57,7 @@ public class MapDBPersistentStoreTest { public void setUp() throws Exception { scheduler = Executors.newScheduledThreadPool(1); - IntegrationUtils.cleanPersistenceFile(BrokerConstants.DEFAULT_PERSISTENT_PATH); + cleanPersistenceFile(BrokerConstants.DEFAULT_PERSISTENT_PATH); Properties props = new Properties(); props.setProperty(BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME, BrokerConstants.DEFAULT_PERSISTENT_PATH); IConfig conf = new MemoryConfig(props); @@ -72,7 +74,17 @@ public class MapDBPersistentStoreTest { } scheduler.shutdown(); - IntegrationUtils.cleanPersistenceFile(BrokerConstants.DEFAULT_PERSISTENT_PATH); + cleanPersistenceFile(BrokerConstants.DEFAULT_PERSISTENT_PATH); + } + + public static void cleanPersistenceFile(String fileName) { + File dbFile = new File(fileName); + if (dbFile.exists()) { + dbFile.delete(); + new File(fileName + ".p").delete(); + new File(fileName + ".t").delete(); + } + assertFalse(dbFile.exists()); } @Test diff --git a/perf/src/main/java/io/moquette/spi/impl/subscriptions/PerformanceTest.java b/perf/src/main/java/io/moquette/spi/impl/subscriptions/PerformanceTest.java index 79a5196a..9db77fcd 100644 --- a/perf/src/main/java/io/moquette/spi/impl/subscriptions/PerformanceTest.java +++ b/perf/src/main/java/io/moquette/spi/impl/subscriptions/PerformanceTest.java @@ -4,8 +4,9 @@ package io.moquette.spi.impl.subscriptions; import java.io.IOException; import java.lang.management.ManagementFactory; import java.lang.management.ThreadMXBean; + +import io.moquette.persistence.MemoryStorageService; import io.moquette.spi.ISessionsStore; -import io.moquette.spi.impl.MemoryStorageService; import io.netty.handler.codec.mqtt.MqttQoS; public class PerformanceTest { diff --git a/pom.xml b/pom.xml index 621309e2..41308562 100644 --- a/pom.xml +++ b/pom.xml @@ -105,6 +105,7 @@ broker + mapdb_storage distribution embedding_moquette osgi_test