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

broker/MapDBSessionStore: code formatter used

Browse Source
This commit is contained in:
Andreas Mager
2017-02-01 17:13:41 +01:00
parent dbefe8e33a
commit f2011e5032
1 changed files with 37 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java
+37 -27
View File
@@ -13,6 +13,7 @@
* *
* You may elect to redistribute this code under either of these licenses. * You may elect to redistribute this code under either of these licenses.
*/ */
package io.moquette.spi.persistence; package io.moquette.spi.persistence;
import io.moquette.spi.ClientSession; import io.moquette.spi.ClientSession;
@@ -25,7 +26,6 @@ import io.moquette.spi.persistence.MapDBPersistentStore.PersistentSession;
import org.mapdb.DB; import org.mapdb.DB;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import java.util.*; import java.util.*;
import java.util.concurrent.BlockingQueue; import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
@@ -40,12 +40,12 @@ class MapDBSessionsStore implements ISessionsStore {
private static final Logger LOG = LoggerFactory.getLogger(MapDBSessionsStore.class); private static final Logger LOG = LoggerFactory.getLogger(MapDBSessionsStore.class);
//maps clientID->[MessageId -> guid] // maps clientID->[MessageId -> guid]
private ConcurrentMap<String, ConcurrentMap<Integer, MessageGUID>> m_inflightStore; private ConcurrentMap<String, ConcurrentMap<Integer, MessageGUID>> m_inflightStore;
//map clientID <-> set of currently in flight packet identifiers // map clientID <-> set of currently in flight packet identifiers
private Map<String, Set<Integer>> m_inFlightIds; private Map<String, Set<Integer>> m_inFlightIds;
private ConcurrentMap<String, PersistentSession> m_persistentSessions; private ConcurrentMap<String, PersistentSession> m_persistentSessions;
//maps clientID->[MessageId -> guid] // maps clientID->[MessageId -> guid]
private ConcurrentMap<String, Map<Integer, MessageGUID>> m_secondPhaseStore; private ConcurrentMap<String, Map<Integer, MessageGUID>> m_secondPhaseStore;
private final DB m_db; private final DB m_db;
@@ -66,14 +66,18 @@ class MapDBSessionsStore implements ISessionsStore {
@Override @Override
public void addNewSubscription(Subscription newSubscription) { public void addNewSubscription(Subscription newSubscription) {
LOG.info("Adding new subscription. CId= {}, topics = {}.", newSubscription.getClientId(), LOG.info(
"Adding new subscription. CId= {}, topics = {}.",
newSubscription.getClientId(),
newSubscription.getTopicFilter()); newSubscription.getTopicFilter());
final String clientID = newSubscription.getClientId(); final String clientID = newSubscription.getClientId();
m_db.getHashMap("subscriptions_" + clientID).put(newSubscription.getTopicFilter(), newSubscription); m_db.getHashMap("subscriptions_" + clientID).put(newSubscription.getTopicFilter(), newSubscription);
if (LOG.isTraceEnabled()) { if (LOG.isTraceEnabled()) {
LOG.trace("The subscription has been added. CId= {}, topics = {}, clientSubscriptions = {}.", LOG.trace(
newSubscription.getClientId(), newSubscription.getTopicFilter(), "The subscription has been added. CId= {}, topics = {}, clientSubscriptions = {}.",
newSubscription.getClientId(),
newSubscription.getTopicFilter(),
m_db.getHashMap("subscriptions_" + clientID)); m_db.getHashMap("subscriptions_" + clientID));
} }
} }
@@ -86,8 +90,11 @@ class MapDBSessionsStore implements ISessionsStore {
} }
m_db.getHashMap("subscriptions_" + clientID).remove(topicFilter); m_db.getHashMap("subscriptions_" + clientID).remove(topicFilter);
if (LOG.isDebugEnabled()) { if (LOG.isDebugEnabled()) {
LOG.debug("The subscription has been removed. CId= {}, topics = {}, clientSubscriptions = {}.", LOG.debug(
clientID, topicFilter, m_db.getHashMap("subscriptions_" + clientID)); "The subscription has been removed. CId= {}, topics = {}, clientSubscriptions = {}.",
clientID,
topicFilter,
m_db.getHashMap("subscriptions_" + clientID));
} }
} }
@@ -96,7 +103,9 @@ class MapDBSessionsStore implements ISessionsStore {
LOG.info("Wiping subscriptions. CId= {}.", clientID); LOG.info("Wiping subscriptions. CId= {}.", clientID);
m_db.delete("subscriptions_" + clientID); m_db.delete("subscriptions_" + clientID);
if (LOG.isDebugEnabled()) { if (LOG.isDebugEnabled()) {
LOG.debug("The subscriptions have been removed. CId= {}, clientSubscriptions = {}.", clientID, LOG.debug(
"The subscriptions have been removed. CId= {}, clientSubscriptions = {}.",
clientID,
m_db.getHashMap("subscriptions_" + clientID)); m_db.getHashMap("subscriptions_" + clientID));
} }
} }
@@ -120,8 +129,7 @@ class MapDBSessionsStore implements ISessionsStore {
@Override @Override
public Subscription getSubscription(ClientTopicCouple couple) { public Subscription getSubscription(ClientTopicCouple couple) {
ConcurrentMap<String, Subscription> clientSubscriptions = m_db.getHashMap("subscriptions_" + couple.clientID); ConcurrentMap<String, Subscription> clientSubscriptions = m_db.getHashMap("subscriptions_" + couple.clientID);
LOG.info("Retrieving subscriptions. CId= {}, subscriptions = {}.", couple.clientID, LOG.info("Retrieving subscriptions. CId= {}, subscriptions = {}.", couple.clientID, clientSubscriptions);
clientSubscriptions);
return clientSubscriptions.get(couple.topicFilter); return clientSubscriptions.get(couple.topicFilter);
} }
@@ -147,7 +155,8 @@ class MapDBSessionsStore implements ISessionsStore {
if (m_persistentSessions.containsKey(clientID)) { if (m_persistentSessions.containsKey(clientID)) {
LOG.error( LOG.error(
"Unable to create a new session: the client ID is already in use. CId= {}, cleanSession = {}.", "Unable to create a new session: the client ID is already in use. CId= {}, cleanSession = {}.",
clientID, cleanSession); clientID,
cleanSession);
throw new IllegalArgumentException("Can't create a session with the ID of an already existing" + clientID); throw new IllegalArgumentException("Can't create a session with the ID of an already existing" + clientID);
} }
LOG.info("Creating new session. CId= {}, cleanSession = {}.", clientID, cleanSession); LOG.info("Creating new session. CId= {}, cleanSession = {}.", clientID, cleanSession);
@@ -166,10 +175,10 @@ class MapDBSessionsStore implements ISessionsStore {
PersistentSession storedSession = m_persistentSessions.get(clientID); PersistentSession storedSession = m_persistentSessions.get(clientID);
return new ClientSession(clientID, m_messagesStore, this, storedSession.cleanSession); return new ClientSession(clientID, m_messagesStore, this, storedSession.cleanSession);
} }
@Override @Override
public Collection<ClientSession> getAllSessions() { public Collection<ClientSession> getAllSessions() {
Collection<ClientSession> result = new ArrayList<ClientSession>(); Collection<ClientSession> result = new ArrayList<>();
for (Map.Entry<String, PersistentSession> entry : m_persistentSessions.entrySet()) { for (Map.Entry<String, PersistentSession> entry : m_persistentSessions.entrySet()) {
result.add(new ClientSession(entry.getKey(), m_messagesStore, this, entry.getValue().cleanSession)); result.add(new ClientSession(entry.getKey(), m_messagesStore, this, entry.getValue().cleanSession));
} }
@@ -184,7 +193,7 @@ class MapDBSessionsStore implements ISessionsStore {
/** /**
* Return the next valid packetIdentifier for the given client session. * Return the next valid packetIdentifier for the given client session.
* */ */
@Override @Override
public int nextPacketID(String clientID) { public int nextPacketID(String clientID) {
if (LOG.isDebugEnabled()) { if (LOG.isDebugEnabled()) {
@@ -219,7 +228,7 @@ class MapDBSessionsStore implements ISessionsStore {
} }
m.remove(messageID); m.remove(messageID);
//remove from the ids store // remove from the ids store
Set<Integer> inFlightForClient = this.m_inFlightIds.get(clientID); Set<Integer> inFlightForClient = this.m_inFlightIds.get(clientID);
if (inFlightForClient != null) { if (inFlightForClient != null) {
inFlightForClient.remove(messageID); inFlightForClient.remove(messageID);
@@ -262,13 +271,14 @@ class MapDBSessionsStore implements ISessionsStore {
} }
MessageGUID guid = m.remove(messageID); MessageGUID guid = m.remove(messageID);
//remove from the ids store // remove from the ids store
Set<Integer> inFlightForClient = this.m_inFlightIds.get(clientID); Set<Integer> inFlightForClient = this.m_inFlightIds.get(clientID);
if (inFlightForClient != null) { if (inFlightForClient != null) {
inFlightForClient.remove(messageID); inFlightForClient.remove(messageID);
} }
Map<Integer, MessageGUID> messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, new HashMap<Integer, MessageGUID>()); Map<Integer, MessageGUID> messageIDs = Utils
.defaultGet(m_secondPhaseStore, clientID, new HashMap<Integer, MessageGUID>());
messageIDs.put(messageID, guid); messageIDs.put(messageID, guid);
m_secondPhaseStore.put(clientID, messageIDs); m_secondPhaseStore.put(clientID, messageIDs);
} }
@@ -278,7 +288,8 @@ class MapDBSessionsStore implements ISessionsStore {
if (LOG.isDebugEnabled()) { if (LOG.isDebugEnabled()) {
LOG.debug("Processing second phase ACK CId={}, messageId={}", clientID, messageID); LOG.debug("Processing second phase ACK CId={}, messageId={}", clientID, messageID);
} }
Map<Integer, MessageGUID> messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, new HashMap<Integer, MessageGUID>()); Map<Integer, MessageGUID> messageIDs = Utils
.defaultGet(m_secondPhaseStore, clientID, new HashMap<Integer, MessageGUID>());
MessageGUID guid = messageIDs.remove(messageID); MessageGUID guid = messageIDs.remove(messageID);
m_secondPhaseStore.put(clientID, messageIDs); m_secondPhaseStore.put(clientID, messageIDs);
return guid; return guid;
@@ -297,22 +308,21 @@ class MapDBSessionsStore implements ISessionsStore {
return "guidsMapping_" + clientID; return "guidsMapping_" + clientID;
} }
@Override @Override
public StoredMessage getInflightMessage(String clientID, int messageID) { public StoredMessage getInflightMessage(String clientID, int messageID) {
LOG.info("Retrieving inflight message CId={}, messageId={}", clientID, messageID); LOG.info("Retrieving inflight message CId={}, messageId={}", clientID, messageID);
Map<Integer, MessageGUID> clientEntries = m_inflightStore.get(clientID); Map<Integer, MessageGUID> clientEntries = m_inflightStore.get(clientID);
if (clientEntries == null) { if (clientEntries == null) {
LOG.warn("The client has no inflight messages CId={}, messageId={}", clientID, messageID); LOG.warn("The client has no inflight messages CId={}, messageId={}", clientID, messageID);
return null; return null;
} }
MessageGUID guid = clientEntries.get(messageID); MessageGUID guid = clientEntries.get(messageID);
if (guid == null) { if (guid == null) {
LOG.warn("The message ID does not have an associated GUID. CId= {}, messageId = {}.", clientID, LOG.warn("The message ID does not have an associated GUID. CId= {}, messageId = {}.", clientID, messageID);
messageID);
return null; return null;
} }
return m_messagesStore.getMessageByGuid(guid); return m_messagesStore.getMessageByGuid(guid);
} }
@Override @Override
public int getInflightMessagesNo(String clientID) { public int getInflightMessagesNo(String clientID) {