anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fixed failing tests after moving to H2 as default store

Browse Source
This commit is contained in:
andsel
2018-11-18 10:58:25 +01:00
parent cfa478067e
commit 1141e9f7a1
20 changed files with 48 additions and 824 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/build.gradle
+1
View File
@@ -33,6 +33,7 @@ dependencies {
testCompile group: 'org.fusesource.mqtt-client', name: 'mqtt-client', version:'1.12'
testCompile group: 'org.eclipse.paho', name: 'org.eclipse.paho.client.mqttv3', version:'1.2.0'
testCompile group: 'org.eclipse.jetty.websocket', name: 'websocket-client', version:'9.2.0.M1'
testCompile group: 'com.h2database', name: 'h2', version:'1.4.191'
testRuntime group: 'io.netty', name: 'netty-tcnative', version: '2.0.10.Final', classifier: 'linux-x86_64'
}
broker/src/main/java/io/moquette/BrokerConstants.java
+2 -2
View File
@@ -27,9 +27,9 @@ public final class BrokerConstants {
public static final String PASSWORD_FILE_PROPERTY_NAME = "password_file";
public static final String PORT_PROPERTY_NAME = "port";
public static final String HOST_PROPERTY_NAME = "host";
public static final String DEFAULT_MOQUETTE_STORE_MAP_DB_FILENAME = "moquette_store.mapdb";
public static final String DEFAULT_MOQUETTE_STORE_H2_DB_FILENAME = "moquette_store.h2";
public static final String DEFAULT_PERSISTENT_PATH = System.getProperty("user.dir") + File.separator
+ DEFAULT_MOQUETTE_STORE_MAP_DB_FILENAME;
+ DEFAULT_MOQUETTE_STORE_H2_DB_FILENAME;
public static final String WEB_SOCKET_PORT_PROPERTY_NAME = "websocket_port";
public static final String WSS_PORT_PROPERTY_NAME = "secure_websocket_port";
broker/src/main/java/io/moquette/broker/Server.java
+8 -1
View File
@@ -52,6 +52,7 @@ public class Server {
private volatile boolean initialized;
private PostOffice dispatcher;
private BrokerInterceptor interceptor;
private H2Builder h2Builder;
public static void main(String[] args) throws IOException {
final Server server = new Server();
@@ -162,7 +163,8 @@ public class Server {
final ISubscriptionsRepository subscriptionsRepository;
if (persistencePath != null && !persistencePath.isEmpty()) {
LOG.trace("Configuring H2 subscriptions store to {}", persistencePath);
subscriptionsRepository = new H2Builder(config, scheduler).initStore().subscriptionsRepository();
h2Builder = new H2Builder(config, scheduler);
subscriptionsRepository = h2Builder.initStore().subscriptionsRepository();
} else {
LOG.trace("Configuring in-memory subscriptions store");
subscriptionsRepository = new MemorySubscriptionsRepository();
@@ -324,6 +326,11 @@ public class Server {
// and SessionsRepository does not stop its tasks. Thus shutdownNow().
scheduler.shutdownNow();
if (h2Builder != null) {
LOG.trace("Shutting down H2 persistence {}");
h2Builder.closeStore();
}
LOG.info("Moquette integration has been stopped.");
}
broker/src/main/java/io/moquette/interception/BrokerInterceptor.java
+1 -1
View File
@@ -160,7 +160,7 @@ public final class BrokerInterceptor implements Interceptor {
public void notifyMessageAcknowledged(final InterceptAcknowledgedMessage msg) {
for (final InterceptHandler handler : this.handlers.get(InterceptAcknowledgedMessage.class)) {
LOG.debug("Notifying MQTT ACK message to interceptor. CId={}, messageId={}, topic={}, interceptorId={}",
msg.getMsg().getClientID(), msg.getPacketID(), msg.getTopic(), handler.getID());
msg.getMsg()/*.getClientID()*/, msg.getPacketID(), msg.getTopic(), handler.getID());
executor.execute(() -> handler.onMessageAcknowledged(msg));
}
}
broker/src/main/java/io/moquette/persistence/H2Builder.java
+4
View File
@@ -48,4 +48,8 @@ public class H2Builder {
public ISubscriptionsRepository subscriptionsRepository() {
return new H2SubscriptionsRepository(mvStore);
}
public void closeStore() {
mvStore.close();
}
}
broker/src/test/java/io/moquette/integration/IntegrationUtils.java
+14 -31
View File
@@ -18,54 +18,37 @@ package io.moquette.integration;
import java.io.File;
import java.util.Properties;
import static io.moquette.BrokerConstants.DEFAULT_MOQUETTE_STORE_MAP_DB_FILENAME;
import static io.moquette.BrokerConstants.DEFAULT_MOQUETTE_STORE_H2_DB_FILENAME;
import static io.moquette.BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME;
import static io.moquette.BrokerConstants.PORT_PROPERTY_NAME;
import static org.junit.Assert.assertFalse;
/**
* Used to carry integration configurations.
*/
public final class IntegrationUtils {
static String localMapDBPath() {
static String localH2MvStoreDBPath() {
String currentDir = System.getProperty("user.dir");
return currentDir + File.separator + "target" + File.separator + DEFAULT_MOQUETTE_STORE_MAP_DB_FILENAME;
}
static String localClusterMapDBPath(int port) {
String currentDir = System.getProperty("user.dir");
return currentDir + File.separator + "target" + File.separator + port + DEFAULT_MOQUETTE_STORE_MAP_DB_FILENAME;
return currentDir + File.separator + "target" + File.separator + DEFAULT_MOQUETTE_STORE_H2_DB_FILENAME;
}
public static Properties prepareTestProperties() {
Properties testProperties = new Properties();
testProperties.put(PERSISTENT_STORE_PROPERTY_NAME, IntegrationUtils.localMapDBPath());
testProperties.put(PERSISTENT_STORE_PROPERTY_NAME, IntegrationUtils.localH2MvStoreDBPath());
testProperties.put(PORT_PROPERTY_NAME, "1883");
return testProperties;
}
public static Properties prepareTestClusterProperties(int port) {
Properties testProperties = new Properties();
testProperties.put(PERSISTENT_STORE_PROPERTY_NAME, IntegrationUtils.localClusterMapDBPath(port));
testProperties.put(PORT_PROPERTY_NAME, Integer.toString(port));
return testProperties;
}
/*public static void cleanPersistenceFile(IConfig config) {
String fileName = config.getProperty(PERSISTENT_STORE_PROPERTY_NAME);
cleanPersistenceFile(fileName);
}
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());
}*/
private IntegrationUtils() {
}
public static void clearTestStorage() {
String dbPath = localH2MvStoreDBPath();
File dbFile = new File(dbPath);
if (dbFile.exists()) {
dbFile.delete();
}
assertFalse(dbFile.exists());
}
}
broker/src/test/java/io/moquette/integration/ServerIntegrationDBAuthenticatorTest.java
+4
View File
@@ -32,6 +32,8 @@ import java.io.IOException;
import java.security.NoSuchAlgorithmException;
import java.sql.SQLException;
import java.util.Properties;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
public class ServerIntegrationDBAuthenticatorTest {
@@ -99,6 +101,8 @@ public class ServerIntegrationDBAuthenticatorTest {
}
stopServer();
IntegrationUtils.clearTestStorage();
}
@AfterClass
broker/src/test/java/io/moquette/integration/ServerIntegrationFuseTest.java
+2
View File
@@ -68,6 +68,8 @@ public class ServerIntegrationFuseTest {
}
m_server.stopServer();
IntegrationUtils.clearTestStorage();
}
@Test
broker/src/test/java/io/moquette/integration/ServerIntegrationOpenSSLTest.java
+1 -1
View File
@@ -55,7 +55,7 @@ public class ServerIntegrationOpenSSLTest extends ServerIntegrationSSLTest {
sslProps.put(BrokerConstants.JKS_PATH_PROPERTY_NAME, "serverkeystore.jks");
sslProps.put(BrokerConstants.KEY_STORE_PASSWORD_PROPERTY_NAME, "passw0rdsrv");
sslProps.put(BrokerConstants.KEY_MANAGER_PASSWORD_PROPERTY_NAME, "passw0rdsrv");
sslProps.put(BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME, IntegrationUtils.localMapDBPath());
sslProps.put(BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME, IntegrationUtils.localH2MvStoreDBPath());
m_server.startServer(sslProps);
}
}
broker/src/test/java/io/moquette/integration/ServerIntegrationPahoTest.java
+2
View File
@@ -80,6 +80,8 @@ public class ServerIntegrationPahoTest {
}
stopServer();
IntegrationUtils.clearTestStorage();
}
private void stopServer() {
broker/src/test/java/io/moquette/integration/ServerIntegrationRestartTest.java
+2
View File
@@ -82,6 +82,8 @@ public class ServerIntegrationRestartTest {
}
m_server.stopServer();
IntegrationUtils.clearTestStorage();
}
@Test
broker/src/test/java/io/moquette/integration/ServerIntegrationSSLClientAuthTest.java
+3 -8
View File
@@ -142,14 +142,14 @@ public class ServerIntegrationSSLClientAuthTest {
sslProps.put(BrokerConstants.JKS_PATH_PROPERTY_NAME, "signedserverkeystore.jks");
sslProps.put(BrokerConstants.KEY_STORE_PASSWORD_PROPERTY_NAME, "passw0rdsrv");
sslProps.put(BrokerConstants.KEY_MANAGER_PASSWORD_PROPERTY_NAME, "passw0rdsrv");
sslProps.put(BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME, IntegrationUtils.localMapDBPath());
sslProps.put(BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME, IntegrationUtils.localH2MvStoreDBPath());
sslProps.put(BrokerConstants.NEED_CLIENT_AUTH, "true");
m_server.startServer(sslProps);
}
@Before
public void setUp() throws Exception {
String dbPath = IntegrationUtils.localMapDBPath();
String dbPath = IntegrationUtils.localH2MvStoreDBPath();
File dbFile = new File(dbPath);
assertFalse(String.format("The DB storagefile %s already exists", dbPath), dbFile.exists());
@@ -171,12 +171,7 @@ public class ServerIntegrationSSLClientAuthTest {
if (m_server != null) {
m_server.stopServer();
}
String dbPath = IntegrationUtils.localMapDBPath();
File dbFile = new File(dbPath);
if (dbFile.exists()) {
dbFile.delete();
}
assertFalse(dbFile.exists());
IntegrationUtils.clearTestStorage();
}
@Test
broker/src/test/java/io/moquette/integration/ServerIntegrationSSLTest.java
+3 -8
View File
@@ -90,13 +90,13 @@ public class ServerIntegrationSSLTest {
sslProps.put(BrokerConstants.JKS_PATH_PROPERTY_NAME, "serverkeystore.jks");
sslProps.put(BrokerConstants.KEY_STORE_PASSWORD_PROPERTY_NAME, "passw0rdsrv");
sslProps.put(BrokerConstants.KEY_MANAGER_PASSWORD_PROPERTY_NAME, "passw0rdsrv");
sslProps.put(BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME, IntegrationUtils.localMapDBPath());
sslProps.put(BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME, IntegrationUtils.localH2MvStoreDBPath());
m_server.startServer(sslProps);
}
@Before
public void setUp() throws Exception {
String dbPath = IntegrationUtils.localMapDBPath();
String dbPath = IntegrationUtils.localH2MvStoreDBPath();
File dbFile = new File(dbPath);
assertFalse(String.format("The DB storagefile %s already exists", dbPath), dbFile.exists());
@@ -118,12 +118,7 @@ public class ServerIntegrationSSLTest {
if (m_server != null) {
m_server.stopServer();
}
String dbPath = IntegrationUtils.localMapDBPath();
File dbFile = new File(dbPath);
if (dbFile.exists()) {
dbFile.delete();
}
assertFalse(dbFile.exists());
IntegrationUtils.clearTestStorage();
}
@Test
distribution/build.gradle
+1 -1
View File
@@ -9,7 +9,7 @@ mainClassName = "io.moquette.integration.Server"
dependencies {
compile project(':moquette-broker')
compile project(':moquette-h2-storage')
compile project(':moquette-mapdb-storage')
//compile project(':moquette-mapdb-storage')
compile group: 'org.slf4j', name: 'slf4j-log4j12', version:'1.7.5'
}
mapdb_storage/build.gradle
-10
View File
@@ -1,10 +0,0 @@
description = 'Moquette - MapDB storage'
dependencies {
compile project(':moquette-broker')
compile group: 'org.mapdb', name: 'mapdb', version:'1.0.8'
testCompile project(path:':moquette-broker', configuration: 'archives')
}
mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBMessagesStore.java
-85
View File
@@ -1,85 +0,0 @@
/*
* Copyright (c) 2012-2018 The original author or authors
* ------------------------------------------------------
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
* and Apache License v2.0 which accompanies this distribution.
*
* The Eclipse Public License is available at
* http://www.eclipse.org/legal/epl-v10.html
*
* The Apache License v2.0 is available at
* http://www.opensource.org/licenses/apache2.0.php
*
* You may elect to redistribute this code under either of these licenses.
*/
package io.moquette.persistence.mapdb;
import io.moquette.spi.IMatchingCondition;
import io.moquette.spi.IMessagesStore;
import io.moquette.broker.subscriptions.Topic;
import org.mapdb.DB;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentMap;
/**
* IMessagesStore implementation backed by MapDB.
*/
class MapDBMessagesStore implements IMessagesStore {
private static final Logger LOG = LoggerFactory.getLogger(MapDBMessagesStore.class);
private DB m_db;
private ConcurrentMap<Topic, StoredMessage> m_retainedStore;
MapDBMessagesStore(DB db) {
m_db = db;
}
@Override
public void initStore() {
m_retainedStore = m_db.getHashMap("retained");
LOG.info("Initialized store");
}
@Override
public Collection<StoredMessage> searchMatching(IMatchingCondition condition) {
LOG.debug("Scanning retained messages");
List<StoredMessage> results = new ArrayList<>();
for (Map.Entry<Topic, StoredMessage> entry : m_retainedStore.entrySet()) {
StoredMessage storedMsg = entry.getValue();
if (condition.match(entry.getKey())) {
results.add(storedMsg);
}
}
if (LOG.isTraceEnabled()) {
LOG.trace("Retained messages have been scanned matchingMessages={}", results);
}
return results;
}
@Override
public void cleanRetained(Topic topic) {
LOG.debug("Cleaning retained messages. Topic={}", topic);
m_retainedStore.remove(topic);
}
@Override
public void storeRetained(Topic topic, StoredMessage storedMessage) {
LOG.debug("Store retained message for topic={}, CId={}", topic, storedMessage.getClientID());
if (storedMessage.getClientID() == null) {
throw new IllegalArgumentException("Message to be persisted must have a not null client ID");
}
m_retainedStore.put(topic, storedMessage);
}
}
mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBPersistentStore.java
-126
View File
@@ -1,126 +0,0 @@
/*
* Copyright (c) 2012-2018 The original author or authors
* ------------------------------------------------------
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
* and Apache License v2.0 which accompanies this distribution.
*
* The Eclipse Public License is available at
* http://www.eclipse.org/legal/epl-v10.html
*
* The Apache License v2.0 is available at
* http://www.opensource.org/licenses/apache2.0.php
*
* You may elect to redistribute this code under either of these licenses.
*/
package io.moquette.persistence.mapdb;
import io.moquette.BrokerConstants;
import io.moquette.broker.config.IConfig;
import io.moquette.spi.IMessagesStore;
import io.moquette.spi.ISessionsStore;
import io.moquette.spi.IStore;
import org.mapdb.DB;
import org.mapdb.DBMaker;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
/**
* MapDB main persistence implementation
*/
public class MapDBPersistentStore implements IStore {
private static final Logger LOG = LoggerFactory.getLogger(MapDBPersistentStore.class);
private DB m_db;
private final String m_storePath;
private final int m_autosaveInterval; // in seconds
protected final ScheduledExecutorService m_scheduler;
private IMessagesStore m_messageStore;
private ISessionsStore m_sessionsStore;
public MapDBPersistentStore(IConfig props, ScheduledExecutorService scheduler) {
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;
}
@Override
public IMessagesStore messagesStore() {
return m_messageStore;
}
@Override
public ISessionsStore sessionsStore() {
return m_sessionsStore;
}
@SuppressWarnings("FutureReturnValueIgnored")
@Override
public void initStore() {
LOG.info("Initializing MapDB store");
if (m_storePath == null || m_storePath.isEmpty()) {
LOG.warn("MapDB store file path is empty, using in-memory store");
m_db = DBMaker.newMemoryDB().make();
} else {
File tmpFile;
try {
LOG.info("Using user-defined MapDB store file. Path={}", m_storePath);
tmpFile = new File(m_storePath);
boolean fileNewlyCreated = tmpFile.createNewFile();
LOG.warn("Using {} MapDB store file. Path={}", fileNewlyCreated ? "fresh" : "existing", m_storePath);
} catch (IOException ex) {
LOG.error("Unable to open MapDB store file. Path={}, cause={}, errorMessage={}", m_storePath,
ex.getCause(), ex.getMessage());
throw new RuntimeException("Can't create temp subscriptions file storage [" + m_storePath + "]", ex);
}
m_db = DBMaker.newFileDB(tmpFile).make();
}
LOG.info("Scheduling MapDB commit task");
m_scheduler.scheduleWithFixedDelay(() -> {
LOG.debug("Committing to MapDB");
m_db.commit();
}, this.m_autosaveInterval, this.m_autosaveInterval, TimeUnit.SECONDS);
// TODO check m_db is valid and
m_messageStore = new MapDBMessagesStore(m_db);
m_messageStore.initStore();
m_sessionsStore = new MapDBSessionsStore(m_db);
m_sessionsStore.initStore();
}
@Override
public void close() {
if (this.m_db.isClosed()) {
LOG.warn("MapDB store is already closed. Nothing will be done");
return;
}
LOG.info("Performing last commit to MapDB");
this.m_db.commit();
LOG.info("Closing MapDB store");
this.m_db.close();
LOG.info("Stopping MapDB commit tasks");
//TODO the 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);
} catch (InterruptedException e) {
}
if (!m_scheduler.isTerminated()) {
LOG.warn("Forcing shutdown of MapDB commit tasks");
m_scheduler.shutdown();
}
LOG.info("MapDB store has been closed successfully");
}
}
mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBSessionsStore.java
-321
View File
@@ -1,321 +0,0 @@
/*
* Copyright (c) 2012-2018 The original author or authors
* ------------------------------------------------------
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
* and Apache License v2.0 which accompanies this distribution.
*
* The Eclipse Public License is available at
* http://www.eclipse.org/legal/epl-v10.html
*
* The Apache License v2.0 is available at
* http://www.opensource.org/licenses/apache2.0.php
*
* You may elect to redistribute this code under either of these licenses.
*/
package io.moquette.persistence.mapdb;
import io.moquette.persistence.PersistentSession;
import io.moquette.spi.IMessagesStore.StoredMessage;
import io.moquette.spi.ISessionsStore;
import io.moquette.spi.ISubscriptionsStore;
import io.moquette.broker.subscriptions.Subscription;
import io.moquette.broker.subscriptions.Topic;
import org.mapdb.DB;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ConcurrentNavigableMap;
/**
* ISessionsStore implementation backed by MapDB.
*/
class MapDBSessionsStore implements ISessionsStore, ISubscriptionsStore {
private static final Logger LOG = LoggerFactory.getLogger(MapDBSessionsStore.class);
// maps clientID->[MessageId -> msg]
private ConcurrentMap<String, ConcurrentMap<Integer, StoredMessage>> outboundFlightMessages;
// map clientID <-> set of currently in flight packet identifiers
private Map<String, Set<Integer>> m_inFlightIds;
private ConcurrentMap<String, PersistentSession> m_persistentSessions;
// maps clientID->[MessageId -> guid]
private ConcurrentMap<String, ConcurrentMap<Integer, StoredMessage>> m_secondPhaseStore;
private final DB m_db;
private ConcurrentNavigableMap<LocalDateTime, Set<String>> sessionsClosingTimes;
MapDBSessionsStore(DB db) {
m_db = db;
}
@Override
public void initStore() {
outboundFlightMessages = m_db.getHashMap("outboundFlight");
m_inFlightIds = m_db.getHashMap("inflightPacketIDs");
m_persistentSessions = m_db.getHashMap("sessions");
m_secondPhaseStore = m_db.getHashMap("secondPhase");
this.sessionsClosingTimes = m_db.getTreeMap("sessionsCreationTimes");
}
@Override
public ISubscriptionsStore subscriptionStore() {
return this;
}
@Override
public void addNewSubscription(Subscription newSubscription) {
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("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. ClientId={}, topics={}", clientID, topicFilter);
if (!m_db.exists("subscriptions_" + clientID)) {
return;
}
m_db.getHashMap("subscriptions_" + clientID).remove(topicFilter);
if (LOG.isDebugEnabled()) {
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);
m_db.delete("subscriptions_" + clientID);
if (LOG.isDebugEnabled()) {
LOG.debug("Subscriptions have been removed. ClientId={}, clientSubscriptions={}", clientID,
m_db.getHashMap("subscriptions_" + clientID));
}
}
@Override
public List<Subscription> listAllSubscriptions() {
LOG.debug("Retrieving existing subscriptions...");
List<Subscription> subscriptions = new ArrayList<>();
for (String clientID : m_persistentSessions.keySet()) {
ConcurrentMap<Topic, Subscription> clientSubscriptions = m_db.getHashMap("subscriptions_" + clientID);
subscriptions.addAll(clientSubscriptions.values());
}
LOG.debug("Existing subscriptions has been retrieved Result={}", subscriptions);
return subscriptions;
}
@Override
public Collection<Subscription> listClientSubscriptions(String clientID) {
ConcurrentMap<Topic, Subscription> clientSubscriptions = m_db.getHashMap("subscriptions_" + clientID);
if (clientSubscriptions == null) {
throw new IllegalStateException("Asking for subscriptions of not persisted client: " + clientID);
}
return clientSubscriptions.values();
}
@Override
public Subscription reload(Subscription subscription) {
final String clientId = subscription.getClientId();
final String clientSubcMapName = "subscriptions_" + clientId;
ConcurrentMap<Topic, Subscription> clientSubscriptions = m_db.getHashMap(clientSubcMapName);
LOG.debug("Retrieving subscriptions. CId={}, subscriptions={}", clientId, clientSubscriptions);
return clientSubscriptions.get(subscription.getTopicFilter());
}
@Override
public boolean contains(String clientID) {
return m_persistentSessions.containsKey(clientID);
}
@Override
public void createNewDurableSession(String clientID) {
m_persistentSessions.putIfAbsent(clientID, new PersistentSession(clientID, false));
}
@Override
public void removeDurableSession(String clientId) {
this.m_persistentSessions.remove(clientId);
this.wipeSubscriptions(clientId);
}
@Override
public void updateCleanStatus(String clientId, boolean newCleanStatus) {
PersistentSession updatedSession = new PersistentSession(clientId, newCleanStatus);
m_persistentSessions.put(clientId, updatedSession);
}
@Override
public PersistentSession loadSessionByKey(String clientID) {
return this.m_persistentSessions.get(clientID);
}
@Override
public Collection<PersistentSession> listAllSessions() {
return this.m_persistentSessions.values();
}
/**
* Return the next valid packetIdentifier for the given client session.
*/
@Override
public int nextPacketID(String clientID) {
LOG.debug("Generating next packet ID CId={}", clientID);
Set<Integer> inFlightForClient = this.m_inFlightIds.get(clientID);
if (inFlightForClient == null) {
int nextPacketId = 1;
inFlightForClient = Collections.newSetFromMap(new ConcurrentHashMap<>());
inFlightForClient.add(nextPacketId);
this.m_inFlightIds.put(clientID, inFlightForClient);
return nextPacketId;
}
int maxId = inFlightForClient.isEmpty() ? 0 : Collections.max(inFlightForClient);
int nextPacketId = (maxId % 0xFFFF) + 1;
inFlightForClient.add(nextPacketId);
LOG.debug("Next packet ID has been generated CId={}, result={}", clientID, nextPacketId);
return nextPacketId;
}
@Override
public StoredMessage inFlightAck(String clientID, int messageID) {
LOG.debug("Acknowledging inflight message CId={}, messageId={}", clientID, messageID);
ConcurrentMap<Integer, StoredMessage> m = this.outboundFlightMessages.get(clientID);
if (m == null) {
LOG.error("Can't find the inFlight record for client <{}>", clientID);
throw new RuntimeException("Can't find the inFlight record for client <" + clientID + ">");
}
StoredMessage msg = m.remove(messageID);
this.outboundFlightMessages.put(clientID, m);
// remove from the ids store
Set<Integer> inFlightForClient = this.m_inFlightIds.get(clientID);
if (inFlightForClient != null) {
inFlightForClient.remove(messageID);
}
return msg;
}
@Override
public void inFlight(String clientID, int messageID, StoredMessage msg) {
ConcurrentMap<Integer, StoredMessage> messages = outboundFlightMessages.get(clientID);
if (messages == null) {
messages = new ConcurrentHashMap<>();
}
messages.put(messageID, msg);
outboundFlightMessages.put(clientID, messages);
}
@Override
public Queue<StoredMessage> queue(String clientID) {
LOG.info("Queuing pending message. CId={}, guid={}", clientID);
return this.m_db.getQueue(clientID);
}
@Override
public void dropQueue(String clientID) {
LOG.info("Removing pending messages. CId={}", clientID);
this.m_db.delete(clientID);
outboundFlightMessages.remove(clientID);
m_inFlightIds.remove(clientID);
}
@Override
public void moveInFlightToSecondPhaseAckWaiting(String clientID, int messageID, StoredMessage msg) {
LOG.debug("Moving inflight message to 2nd phase ack state. CId={}, messageID={}", clientID, messageID);
ConcurrentMap<Integer, StoredMessage> m = this.m_secondPhaseStore.get(clientID);
if (m == null) {
String error = String.format("Can't find the inFlight record for client <%s> during the second phase of " +
"QoS2 pub", clientID);
LOG.error(error);
throw new RuntimeException(error);
}
m.put(messageID, msg);
this.outboundFlightMessages.put(clientID, m);
}
@Override
public StoredMessage completeReleasedPublish(String clientID, int messageID) {
LOG.debug("Processing second phase ACK CId={}, messageId={}", clientID, messageID);
final ConcurrentMap<Integer, StoredMessage> m = this.m_secondPhaseStore.get(clientID);
if (m == null) {
String error = String.format("Can't find the inFlight record for client <%s> during the second phase " +
"acking of QoS2 pub", clientID);
LOG.error(error);
throw new RuntimeException(error);
}
StoredMessage msg = m.remove(messageID);
m_secondPhaseStore.put(clientID, m);
return msg;
}
@Override
public int getInflightMessagesNo(String clientID) {
int totalInflight = 0;
final String infligthMessageMapName = inboundMessageId2MessagesMapName(clientID);
ConcurrentMap<Integer, StoredMessage> inflightPerClient = m_db.getHashMap(infligthMessageMapName);
if (inflightPerClient != null) {
totalInflight += inflightPerClient.size();
}
// totalInflight += countPubReleaseWaitingPubComplete(clientID);
Map<Integer, StoredMessage> outboundPerClient = outboundFlightMessages.get(clientID);
if (outboundPerClient != null) {
totalInflight += outboundPerClient.size();
}
return totalInflight;
}
@Override
public int countPubReleaseWaitingPubComplete(String clientID) {
if (!m_secondPhaseStore.containsKey(clientID))
return 0;
return m_secondPhaseStore.get(clientID).size();
}
@Override
public void removeTemporaryQoS2(String clientID) {
LOG.info("Removing stored messages with QoS 2. ClientId={}", clientID);
m_secondPhaseStore.remove(clientID);
}
private static String inboundMessageId2MessagesMapName(String clientID) {
return "inboundInflight_" + clientID;
}
@Override
public synchronized void trackSessionClose(LocalDateTime when, String clientID) {
this.sessionsClosingTimes.putIfAbsent(when, new HashSet<>());
this.sessionsClosingTimes.computeIfPresent(when, (key, oldSet) -> {
oldSet.add(clientID);
return oldSet;
});
}
@Override
public Set<String> sessionOlderThan(LocalDateTime queryPin) {
final Set<String> results = new HashSet<>();
LocalDateTime keyBefore = this.sessionsClosingTimes.lowerKey(queryPin);
while (keyBefore != null) {
results.addAll(this.sessionsClosingTimes.get(keyBefore));
keyBefore = this.sessionsClosingTimes.lowerKey(keyBefore);
}
return results;
}
}
mapdb_storage/src/test/java/io/moquette/persistence/mapdb/MapDBPersistentStoreTest.java
-108
View File
@@ -1,108 +0,0 @@
/*
* Copyright (c) 2012-2018 The original author or authors
* ------------------------------------------------------
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
* and Apache License v2.0 which accompanies this distribution.
*
* The Eclipse Public License is available at
* http://www.eclipse.org/legal/epl-v10.html
*
* The Apache License v2.0 is available at
* http://www.opensource.org/licenses/apache2.0.php
*
* You may elect to redistribute this code under either of these licenses.
*/
package io.moquette.persistence.mapdb;
/**
*
* @author andrea
*/
public class MapDBPersistentStoreTest /*extends MessageStoreTCK*/ {
// MapDBPersistentStore m_storageService;
//
// private ScheduledExecutorService scheduler;
//
// @Before
// public void setUp() throws Exception {
// scheduler = Executors.newScheduledThreadPool(1);
//
// 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);
// m_storageService = new MapDBPersistentStore(conf, scheduler);
// m_storageService.initStore();
// messagesStore = m_storageService.messagesStore();
// sessionsStore = m_storageService.sessionsStore();
// this.sessionsRepository = new SessionsRepository(sessionsStore, null);
// }
//
// @After
// public void tearDown() {
// if (m_storageService != null) {
// m_storageService.close();
// }
//
// scheduler.shutdown();
// cleanPersistenceFile(BrokerConstants.DEFAULT_PERSISTENT_PATH);
// }
//
// 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
// public void testNextPacketID_notExistingClientSession() {
// int packetId = sessionsStore.nextPacketID("NOT_EXISTING_CLI");
// assertEquals(1, packetId);
// }
//
// @Test
// public void testNextPacketID_existingClientSession() {
// // Force creation of inflight map for the CLIENT session
// int packetId = sessionsStore.nextPacketID("CLIENT");
// assertEquals(1, packetId);
//
// // request a second packetID
// packetId = sessionsStore.nextPacketID("CLIENT");
// assertEquals(2, packetId);
// }
//
// @Test
// public void testNextPacketID() {
// StoredMessage msgStored = new StoredMessage("Hello".getBytes(UTF_8), MqttQoS.AT_LEAST_ONCE, "/topic");
// msgStored.setClientID(TEST_CLIENT);
//
// // request a first ID
// int packetId = sessionsStore.nextPacketID("CLIENT");
// sessionsStore.inFlight("CLIENT", packetId, msgStored); // simulate an inflight
// assertEquals(1, packetId);
//
// // release the ID
// sessionsStore.inFlightAck("CLIENT", packetId);
//
// // request a second packetID, counter restarts from 0
// packetId = sessionsStore.nextPacketID("CLIENT");
// assertEquals(1, packetId);
// }
//
// @Test
// public void testCloseShutdownCommitTask() throws InterruptedException {
// m_storageService.close();
//
// // verify the executor is shutdown
// assertTrue("Storage service scheduler can't be stopped in 3 seconds",
// m_storageService.m_scheduler.awaitTermination(3, TimeUnit.SECONDS));
// assertTrue(m_storageService.m_scheduler.isTerminated());
// }
}
mapdb_storage/src/test/java/io/moquette/persistence/mapdb/MapDBSessionsStoreTest.java
-121
View File
@@ -1,121 +0,0 @@
/*
* Copyright (c) 2012-2018 The original author or authors
* ------------------------------------------------------
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
* and Apache License v2.0 which accompanies this distribution.
*
* The Eclipse Public License is available at
* http://www.eclipse.org/legal/epl-v10.html
*
* The Apache License v2.0 is available at
* http://www.opensource.org/licenses/apache2.0.php
*
* You may elect to redistribute this code under either of these licenses.
*/
package io.moquette.persistence.mapdb;
//import static io.moquette.persistence.mapdb.MapDBPersistentStoreTest.cleanPersistenceFile;
import static java.time.LocalDateTime.now;
import static org.assertj.core.api.Assertions.assertThat;
public class MapDBSessionsStoreTest {
// private static final Logger LOG = LoggerFactory.getLogger(MapDBSessionsStoreTest.class);
//
// MapDBPersistentStore storageService;
//
// private ScheduledExecutorService scheduler;
//
// @Before
// public void setUp() throws Exception {
// scheduler = Executors.newScheduledThreadPool(1);
//
// 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);
// storageService = new MapDBPersistentStore(conf, scheduler);
// storageService.initStore();
// }
//
// @After
// public void tearDown() {
// if (storageService != null) {
// storageService.close();
// }
//
// scheduler.shutdown();
// cleanPersistenceFile(BrokerConstants.DEFAULT_PERSISTENT_PATH);
// }
//
// @Test
// public void testQueryByExpirationDate() {
// MapDBSessionsStore sessionsStore = (MapDBSessionsStore) storageService.sessionsStore();
// LocalDateTime sessionCreationTime = LocalDateTime.of(2017, 10, 1, 10, 0, 0);
// sessionsStore.trackSessionClose(sessionCreationTime, "Sensor1");
//
// LocalDateTime session2CreationTime = LocalDateTime.of(2017, 11, 1, 10, 0, 0);
// sessionsStore.trackSessionClose(session2CreationTime, "Sensor2");
//
// //Exercise
// LocalDateTime queryPin = LocalDateTime.of(2017, 10, 20, 10, 0, 0);
// Set<String> sessions = sessionsStore.sessionOlderThan(queryPin);
//
// //Verify
// assertThat(sessions).contains("Sensor1");
// }
//
// @Test
// public void testListAllSessionsOlderThan() {
// MapDBSessionsStore sessionsStore = (MapDBSessionsStore) storageService.sessionsStore();
// LocalDateTime sessionCreationTime = LocalDateTime.of(2017, 10, 1, 10, 0, 0);
// sessionsStore.trackSessionClose(sessionCreationTime, "Sensor1");
//
// LocalDateTime session2CreationTime = LocalDateTime.of(2017, 10, 2, 10, 0, 0);
// sessionsStore.trackSessionClose(session2CreationTime, "Sensor2");
//
// //Exercise
// LocalDateTime queryPin = LocalDateTime.of(2017, 10, 20, 10, 0, 0);
// Set<String> sessions = sessionsStore.sessionOlderThan(queryPin);
//
// //Verify
// assertThat(sessions).contains("Sensor1", "Sensor2");
// }
//
// @Test
// public void testRetrieve1000ExpiredSessionPerformance() throws InterruptedException, ExecutionException,
// TimeoutException {
// MapDBSessionsStore sessionsStore = (MapDBSessionsStore) storageService.sessionsStore();
//
// final ExecutorService executorService = Executors.newFixedThreadPool(2);
// final Future<?> future1 = executorService.submit(inserterForOffset(sessionsStore, 0));
// final Future<?> future2 = executorService.submit(inserterForOffset(sessionsStore, 5000));
// future1.get(10, TimeUnit.SECONDS);
// future2.get(10, TimeUnit.SECONDS);
// //prepare the fixture
//// for (int i = 0; i < 100000; i++) {
//// final LocalDateTime timePin = now().minusSeconds(1);
//// sessionsStore.trackSessionClose(timePin, "Sensor" + i);
//// }
//
// //Exercise
// LOG.warn("Start search");
// Set<String> sessions = sessionsStore.sessionOlderThan(now());
// LOG.warn("Finish search");
//
// //Verify
// assertThat(sessions.size()).isEqualTo(10000);
// }
//
// private Runnable inserterForOffset(MapDBSessionsStore sessionsStore, int start) {
// return () -> {
// //prepare the fixture
// for (int i = start; i < start + 5000; i++) {
// final LocalDateTime timePin = now().minusSeconds(1);
// sessionsStore.trackSessionClose(timePin, "Sensor" + i);
// }
// };
// }
}