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

Changed String for guids to its value object

Browse Source
This commit is contained in:
andsel
2016-09-23 11:51:25 +02:00
parent ac419adbd6
commit 8215e89d77
12 changed files with 174 additions and 117 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/spi/ClientSession.java
+6 -6
View File
@@ -78,14 +78,14 @@ public class ClientSession {
* */
public List<IMessagesStore.StoredMessage> storedMessages() {
//read all messages from enqueued store
Collection<String> guids = this.m_sessionsStore.enqueued(clientID);
Collection<MessageGUID> guids = this.m_sessionsStore.enqueued(clientID);
return messagesStore.listMessagesInSession(guids);
}
/**
* Remove the messages stored in a cleanSession false.
*/
public void removeEnqueued(String guid) {
public void removeEnqueued(MessageGUID guid) {
this.m_sessionsStore.removeEnqueued(this.clientID, guid);
}
@@ -175,22 +175,22 @@ public class ClientSession {
m_sessionsStore.inFlightAck(this.clientID, messageID);
}
public void inFlightAckWaiting(String guid, int messageID) {
public void inFlightAckWaiting(MessageGUID guid, int messageID) {
LOG.trace("Adding to inflight {}, guid <{}>", messageID, guid);
m_sessionsStore.inFlight(this.clientID, messageID, guid);
}
public IMessagesStore.StoredMessage secondPhaseAcknowledged(int messageID) {
String guid = m_sessionsStore.secondPhaseAcknowledged(clientID, messageID);
MessageGUID guid = m_sessionsStore.secondPhaseAcknowledged(clientID, messageID);
return messagesStore.getMessageByGuid(guid);
}
public void enqueueToDeliver(String guid) {
public void enqueueToDeliver(MessageGUID guid) {
this.m_sessionsStore.bindToDeliver(guid, this.clientID);
}
public IMessagesStore.StoredMessage storedMessage(int messageID) {
final String guid = m_sessionsStore.mapToGuid(clientID, messageID);
final MessageGUID guid = m_sessionsStore.mapToGuid(clientID, messageID);
return messagesStore.getMessageByGuid(guid);
}
broker/src/main/java/io/moquette/spi/IMessagesStore.java
+9 -9
View File
@@ -35,7 +35,7 @@ public interface IMessagesStore {
private String m_clientID;
//Optional attribute, available only fo QoS 1 and 2
private Integer m_msgID;
private String m_guid;
private MessageGUID m_guid;
private int referenceCounter = 0;
public StoredMessage(byte[] message, AbstractMessage.QOSType qos, String topic) {
@@ -56,11 +56,11 @@ public interface IMessagesStore {
return m_topic;
}
public void setGuid(String guid) {
public void setGuid(MessageGUID guid) {
this.m_guid = guid;
}
public String getGuid() {
public MessageGUID getGuid() {
return m_guid;
}
@@ -125,7 +125,7 @@ public interface IMessagesStore {
* Persist the message.
* If the message is empty then the topic is cleaned, else it's stored.
*/
void storeRetained(String topic, String guid);
void storeRetained(String topic, MessageGUID guid);
/**
* Return a list of retained messages that satisfy the condition.
@@ -136,22 +136,22 @@ public interface IMessagesStore {
* Persist the message.
* @return the unique id in the storage (guid).
* */
String storePublishForFuture(StoredMessage evt);
MessageGUID storePublishForFuture(StoredMessage evt);
/**
* Return the list of persisted publishes for the given clientID.
* For QoS1 and QoS2 with clean session flag, this method return the list of
* missed publish events while the client was disconnected.
*/
List<StoredMessage> listMessagesInSession(Collection<String> guids);
List<StoredMessage> listMessagesInSession(Collection<MessageGUID> guids);
void dropMessagesInSession(String clientID);
StoredMessage getMessageByGuid(String guid);
StoredMessage getMessageByGuid(MessageGUID guid);
void cleanRetained(String topic);
void incUsageCounter(String guid);
void incUsageCounter(MessageGUID guid);
void decUsageCounter(String guid);
void decUsageCounter(MessageGUID guid);
}
broker/src/main/java/io/moquette/spi/ISessionsStore.java
+6 -6
View File
@@ -112,7 +112,7 @@ public interface ISessionsStore {
/**
* Save the binding messageID, clientID <-> guid
* */
void inFlight(String clientID, int messageID, String guid);
void inFlight(String clientID, int messageID, MessageGUID guid);
/**
* Return the next valid packetIdentifier for the given client session.
@@ -122,26 +122,26 @@ public interface ISessionsStore {
/**
* Store the guid to be later published.
* */
void bindToDeliver(String guid, String clientID);
void bindToDeliver(MessageGUID guid, String clientID);
/**
* List the guids for retained messages for the session
* */
Collection<String> enqueued(String clientID);
Collection<MessageGUID> enqueued(String clientID);
/**
* Remove form the queue of stored messages for session.
* */
void removeEnqueued(String clientID, String guid);
void removeEnqueued(String clientID, MessageGUID guid);
void moveInFlightToSecondPhaseAckWaiting(String clientID, int messageID);
/**
* @return the guid of message just acked.
* */
String secondPhaseAcknowledged(String clientID, int messageID);
MessageGUID secondPhaseAcknowledged(String clientID, int messageID);
String mapToGuid(String clientID, int messageID);
MessageGUID mapToGuid(String clientID, int messageID);
StoredMessage getInflightMessage(String clientID, int messageID);
}
broker/src/main/java/io/moquette/spi/MessageGUID.java
+52
View File
@@ -0,0 +1,52 @@
/*
* Copyright (c) 2012-2015 The original author or authors
* ------------------------------------------------------
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
* and Apache License v2.0 which accompanies this distribution.
*
* The Eclipse Public License is available at
* http://www.eclipse.org/legal/epl-v10.html
*
* The Apache License v2.0 is available at
* http://www.opensource.org/licenses/apache2.0.php
*
* You may elect to redistribute this code under either of these licenses.
*/
package io.moquette.spi;
import java.io.Serializable;
/**
* Value object for GUIDs of messages.
* */
public class MessageGUID implements Serializable {
private final String guid;
public MessageGUID(String guid) {
this.guid = guid;
}
@Override
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
MessageGUID that = (MessageGUID) o;
return guid != null ? guid.equals(that.guid) : that.guid == null;
}
@Override
public int hashCode() {
return guid != null ? guid.hashCode() : 0;
}
@Override
public String toString() {
return "MessageGUID{" +
"guid='" + guid + '\'' +
'}';
}
}
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+4 -8
View File
@@ -25,11 +25,8 @@ import io.moquette.interception.InterceptHandler;
import io.moquette.server.ConnectionDescriptor;
import io.moquette.server.netty.AutoFlushHandler;
import io.moquette.server.netty.NettyUtils;
import io.moquette.spi.ClientSession;
import io.moquette.spi.IMatchingCondition;
import io.moquette.spi.IMessagesStore;
import io.moquette.spi.*;
import io.moquette.spi.IMessagesStore.StoredMessage;
import io.moquette.spi.ISessionsStore;
import io.moquette.spi.security.IAuthenticator;
import io.moquette.spi.security.IAuthorizator;
import io.moquette.spi.impl.subscriptions.SubscriptionsStore;
@@ -150,7 +147,6 @@ public class ProtocolProcessor {
* @param allowZeroByteClientId true to allow clients connect without a clientid
* @param authorizator used to apply ACL policies to publishes and subscriptions.
* @param interceptor to notify events to an intercept handler
* @param serverPort
*/
void init(SubscriptionsStore subscriptions, IMessagesStore storageService,
ISessionsStore sessionsStore,
@@ -387,7 +383,7 @@ public class ProtocolProcessor {
final Integer messageID = msg.getMessageID();
LOG.info("PUBLISH on server {} from clientID <{}> on topic <{}> with QoS {}", m_server_port, clientID, topic, qos);
String guid = null;
MessageGUID guid = null;
IMessagesStore.StoredMessage toStoreMsg = asStoredMessage(msg);
toStoreMsg.setClientID(clientID);
if (qos == AbstractMessage.QOSType.MOST_ONE) { //QoS0
@@ -439,7 +435,7 @@ public class ProtocolProcessor {
final String topic = msg.getTopicName();
LOG.info("embedded PUBLISH on topic <{}> with QoS {}", topic, qos);
String guid = null;
MessageGUID guid = null;
IMessagesStore.StoredMessage toStoreMsg = asStoredMessage(msg);
if (msg.getClientId() == null || msg.getClientId().isEmpty()) {
toStoreMsg.setClientID("BROKER_SELF");
@@ -498,7 +494,7 @@ public class ProtocolProcessor {
LOG.trace("subscription tree {}", subscriptions.dumpTree());
}
//if QoS 1 or 2 store the message
String guid = null;
MessageGUID guid = null;
if (publishingQos == QOSType.EXACTLY_ONCE || publishingQos == QOSType.LEAST_ONE) {
guid = m_messagesStore.storePublishForFuture(pubMsg);
}
broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java
+17 -16
View File
@@ -17,6 +17,7 @@ package io.moquette.spi.persistence;
import io.moquette.spi.IMatchingCondition;
import io.moquette.spi.IMessagesStore;
import io.moquette.spi.MessageGUID;
import org.mapdb.DB;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -36,9 +37,9 @@ class MapDBMessagesStore implements IMessagesStore {
private DB m_db;
//maps clientID -> guid
private ConcurrentMap<String, String> m_retainedStore;
private ConcurrentMap<String, MessageGUID> m_retainedStore;
//maps guid to message, it's message store
private ConcurrentMap<String, IMessagesStore.StoredMessage> m_persistentMessageStore;
private ConcurrentMap<MessageGUID, IMessagesStore.StoredMessage> m_persistentMessageStore;
MapDBMessagesStore(DB db) {
@@ -52,7 +53,7 @@ class MapDBMessagesStore implements IMessagesStore {
}
@Override
public void storeRetained(String topic, String guid) {
public void storeRetained(String topic, MessageGUID guid) {
m_retainedStore.put(topic, guid);
}
@@ -61,8 +62,8 @@ class MapDBMessagesStore implements IMessagesStore {
LOG.debug("searchMatching scanning all retained messages, presents are {}", m_retainedStore.size());
List<StoredMessage> results = new ArrayList<>();
for (Map.Entry<String, String> entry : m_retainedStore.entrySet()) {
final String guid = entry.getValue();
for (Map.Entry<String, MessageGUID> entry : m_retainedStore.entrySet()) {
final MessageGUID guid = entry.getValue();
StoredMessage storedMsg = m_persistentMessageStore.get(guid);
if (condition.match(entry.getKey())) {
results.add(storedMsg);
@@ -73,25 +74,25 @@ class MapDBMessagesStore implements IMessagesStore {
}
@Override
public String storePublishForFuture(StoredMessage evt) {
public MessageGUID storePublishForFuture(StoredMessage evt) {
LOG.debug("storePublishForFuture store evt {}", evt);
if (evt.getClientID() == null) {
LOG.error("persisting a message without a clientID, bad programming error msg: {}", evt);
throw new IllegalArgumentException("\"persisting a message without a clientID, bad programming error");
}
String guid = UUID.randomUUID().toString();
MessageGUID guid = new MessageGUID(UUID.randomUUID().toString());
evt.setGuid(guid);
LOG.debug("storePublishForFuture guid <{}>", guid);
m_persistentMessageStore.put(guid, evt);
ConcurrentMap<Integer, String> messageIdToGuid = m_db.getHashMap(MapDBSessionsStore.messageId2GuidsMapName(evt.getClientID()));
ConcurrentMap<Integer, MessageGUID> messageIdToGuid = m_db.getHashMap(MapDBSessionsStore.messageId2GuidsMapName(evt.getClientID()));
messageIdToGuid.put(evt.getMessageID(), guid);
return guid;
}
@Override
public List<StoredMessage> listMessagesInSession(Collection<String> guids) {
public List<StoredMessage> listMessagesInSession(Collection<MessageGUID> guids) {
List<StoredMessage> ret = new ArrayList<>();
for (String guid : guids) {
for (MessageGUID guid : guids) {
ret.add(m_persistentMessageStore.get(guid));
}
return ret;
@@ -99,14 +100,14 @@ class MapDBMessagesStore implements IMessagesStore {
@Override
public void dropMessagesInSession(String clientID) {
ConcurrentMap<Integer, String> messageIdToGuid = m_db.getHashMap(MapDBSessionsStore.messageId2GuidsMapName(clientID));
for (String guid : messageIdToGuid.values()) {
ConcurrentMap<Integer, MessageGUID> messageIdToGuid = m_db.getHashMap(MapDBSessionsStore.messageId2GuidsMapName(clientID));
for (MessageGUID guid : messageIdToGuid.values()) {
removeStoredMessage(guid);
}
messageIdToGuid.clear();
}
void removeStoredMessage(String guid) {
void removeStoredMessage(MessageGUID guid) {
//remove only the not retained and no more referenced
StoredMessage storedMessage = m_persistentMessageStore.get(guid);
if (!storedMessage.isRetained() && storedMessage.getReferenceCounter() == 0) {
@@ -116,7 +117,7 @@ class MapDBMessagesStore implements IMessagesStore {
}
@Override
public StoredMessage getMessageByGuid(String guid) {
public StoredMessage getMessageByGuid(MessageGUID guid) {
return m_persistentMessageStore.get(guid);
}
@@ -126,14 +127,14 @@ class MapDBMessagesStore implements IMessagesStore {
}
@Override
public void incUsageCounter(String guid) {
public void incUsageCounter(MessageGUID guid) {
IMessagesStore.StoredMessage storedMessage = m_persistentMessageStore.get(guid);
storedMessage.incReferenceCounter();
m_persistentMessageStore.replace(guid, storedMessage);
}
@Override
public void decUsageCounter(String guid) {
public void decUsageCounter(MessageGUID guid) {
IMessagesStore.StoredMessage storedMessage = m_persistentMessageStore.get(guid);
storedMessage.decReferenceCounter();
m_persistentMessageStore.replace(guid, storedMessage);
broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java
+23 -22
View File
@@ -18,6 +18,7 @@ package io.moquette.spi.persistence;
import io.moquette.spi.ClientSession;
import io.moquette.spi.IMessagesStore.StoredMessage;
import io.moquette.spi.ISessionsStore;
import io.moquette.spi.MessageGUID;
import io.moquette.spi.impl.Utils;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.spi.persistence.MapDBPersistentStore.PersistentSession;
@@ -38,14 +39,14 @@ class MapDBSessionsStore implements ISessionsStore {
private static final Logger LOG = LoggerFactory.getLogger(MapDBSessionsStore.class);
//maps clientID->[MessageId -> guid]
private ConcurrentMap<String, Map<Integer, String>> m_inflightStore;
private ConcurrentMap<String, Map<Integer, MessageGUID>> m_inflightStore;
//map clientID <-> set of currently in flight packet identifiers
private Map<String, Set<Integer>> m_inFlightIds;
private ConcurrentMap<String, PersistentSession> m_persistentSessions;
//maps clientID->[guid*], insertion order cares, it's queue
private ConcurrentMap<String, List<String>> m_enqueuedStore;
private ConcurrentMap<String, List<MessageGUID>> m_enqueuedStore;
//maps clientID->[MessageId -> guid]
private ConcurrentMap<String, Map<Integer, String>> m_secondPhaseStore;
private ConcurrentMap<String, Map<Integer, MessageGUID>> m_secondPhaseStore;
private final DB m_db;
private final MapDBMessagesStore m_messagesStore;
@@ -172,7 +173,7 @@ class MapDBSessionsStore implements ISessionsStore {
@Override
public void inFlightAck(String clientID, int messageID) {
LOG.info("acknowledging inflight clientID <{}> messageID {}", clientID, messageID);
Map<Integer, String> m = this.m_inflightStore.get(clientID);
Map<Integer, MessageGUID> m = this.m_inflightStore.get(clientID);
if (m == null) {
LOG.error("Can't find the inFlight record for client <{}>", clientID);
return;
@@ -187,8 +188,8 @@ class MapDBSessionsStore implements ISessionsStore {
}
@Override
public void inFlight(String clientID, int messageID, String guid) {
Map<Integer, String> m = this.m_inflightStore.get(clientID);
public void inFlight(String clientID, int messageID, MessageGUID guid) {
Map<Integer, MessageGUID> m = this.m_inflightStore.get(clientID);
if (m == null) {
m = new HashMap<>();
}
@@ -198,21 +199,21 @@ class MapDBSessionsStore implements ISessionsStore {
}
@Override
public void bindToDeliver(String guid, String clientID) {
List<String> guids = Utils.defaultGet(m_enqueuedStore, clientID, new ArrayList<String>());
public void bindToDeliver(MessageGUID guid, String clientID) {
List<MessageGUID> guids = Utils.defaultGet(m_enqueuedStore, clientID, new ArrayList<MessageGUID>());
guids.add(guid);
m_enqueuedStore.put(clientID, guids);
m_messagesStore.incUsageCounter(guid);
}
@Override
public Collection<String> enqueued(String clientID) {
return Utils.defaultGet(m_enqueuedStore, clientID, new ArrayList<String>());
public Collection<MessageGUID> enqueued(String clientID) {
return Utils.defaultGet(m_enqueuedStore, clientID, new ArrayList<MessageGUID>());
}
@Override
public void removeEnqueued(String clientID, String guid) {
List<String> guids = Utils.defaultGet(m_enqueuedStore, clientID, new ArrayList<String>());
public void removeEnqueued(String clientID, MessageGUID guid) {
List<MessageGUID> guids = Utils.defaultGet(m_enqueuedStore, clientID, new ArrayList<MessageGUID>());
guids.remove(guid);
m_enqueuedStore.put(clientID, guids);
m_messagesStore.decUsageCounter(guid);
@@ -223,12 +224,12 @@ class MapDBSessionsStore implements ISessionsStore {
@Override
public void moveInFlightToSecondPhaseAckWaiting(String clientID, int messageID) {
LOG.info("acknowledging inflight clientID <{}> messageID {}", clientID, messageID);
Map<Integer, String> m = this.m_inflightStore.get(clientID);
Map<Integer, MessageGUID> m = this.m_inflightStore.get(clientID);
if (m == null) {
LOG.error("Can't find the inFlight record for client <{}>", clientID);
return;
}
String guid = m.remove(messageID);
MessageGUID guid = m.remove(messageID);
//remove from the ids store
Set<Integer> inFlightForClient = this.m_inFlightIds.get(clientID);
@@ -237,22 +238,22 @@ class MapDBSessionsStore implements ISessionsStore {
}
LOG.info("Moving to second phase store");
Map<Integer, String> messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, new HashMap<Integer, String>());
Map<Integer, MessageGUID> messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, new HashMap<Integer, MessageGUID>());
messageIDs.put(messageID, guid);
m_secondPhaseStore.put(clientID, messageIDs);
}
@Override
public String secondPhaseAcknowledged(String clientID, int messageID) {
Map<Integer, String> messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, new HashMap<Integer, String>());
String guid = messageIDs.remove(messageID);
public MessageGUID secondPhaseAcknowledged(String clientID, int messageID) {
Map<Integer, MessageGUID> messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, new HashMap<Integer, MessageGUID>());
MessageGUID guid = messageIDs.remove(messageID);
m_secondPhaseStore.put(clientID, messageIDs);
return guid;
}
@Override
public String mapToGuid(String clientID, int messageID) {
ConcurrentMap<Integer, String> messageIdToGuid = m_db.getHashMap(messageId2GuidsMapName(clientID));
public MessageGUID mapToGuid(String clientID, int messageID) {
ConcurrentMap<Integer, MessageGUID> messageIdToGuid = m_db.getHashMap(messageId2GuidsMapName(clientID));
return messageIdToGuid.get(messageID);
}
@@ -262,10 +263,10 @@ class MapDBSessionsStore implements ISessionsStore {
@Override
public StoredMessage getInflightMessage(String clientID, int messageID) {
Map<Integer, String> clientEntries = m_inflightStore.get(clientID);
Map<Integer, MessageGUID> clientEntries = m_inflightStore.get(clientID);
if (clientEntries == null)
return null;
String guid = clientEntries.get(messageID);
MessageGUID guid = clientEntries.get(messageID);
LOG.info("inflight messageID {} guid <{}>", messageID, guid);
if (guid == null)
return null;