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

Changed retained persistence to use the global message store and guids, instead of duplicate msg contents

Browse Source
This commit is contained in:
andsel
2015-11-18 22:25:14 +01:00
parent faf1abb896
commit 9f9d01ec58
8 changed files with 155 additions and 278 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/eclipse/moquette/spi/ClientSession.java
+2 -3
View File
@@ -16,7 +16,6 @@
package org.eclipse.moquette.spi;
import org.eclipse.moquette.proto.messages.AbstractMessage;
import org.eclipse.moquette.spi.impl.events.PublishEvent;
import org.eclipse.moquette.spi.impl.subscriptions.Subscription;
import org.eclipse.moquette.spi.impl.subscriptions.SubscriptionsStore;
import org.slf4j.Logger;
@@ -73,7 +72,7 @@ public class ClientSession {
/**
* @return the list of messages to be delivered for client related to the session.
* */
public List<PublishEvent> storedMessages() {
public List<IMessagesStore.StoredMessage> storedMessages() {
//read all messages from enqueued store
Collection<String> guids = this.m_sessionsStore.enqueued(clientID);
return messagesStore.listMessagesInSession(guids);
@@ -172,7 +171,7 @@ public class ClientSession {
this.m_sessionsStore.bindToDeliver(guid, this.clientID);
}
public PublishEvent storedMessage(int messageID) {
public IMessagesStore.StoredMessage storedMessage(int messageID) {
final String guid = m_sessionsStore.mapToGuid(clientID, messageID);
return messagesStore.getMessageByGuid(guid);
}
broker/src/main/java/org/eclipse/moquette/spi/IMessagesStore.java
+56 -5
View File
@@ -17,7 +17,6 @@ package org.eclipse.moquette.spi;
import java.io.Serializable;
import java.nio.ByteBuffer;
import org.eclipse.moquette.spi.impl.events.PublishEvent;
import org.eclipse.moquette.proto.messages.AbstractMessage;
import java.util.Collection;
@@ -32,6 +31,11 @@ public interface IMessagesStore {
final AbstractMessage.QOSType m_qos;
final byte[] m_payload;
final String m_topic;
private boolean m_retained;
private String m_clientID;
//Optional attribute, available only fo QoS 1 and 2
private Integer m_msgID;
private String m_guid;
public StoredMessage(byte[] message, AbstractMessage.QOSType qos, String topic) {
m_qos = qos;
@@ -50,6 +54,53 @@ public interface IMessagesStore {
public String getTopic() {
return m_topic;
}
public void setGuid(String guid) {
this.m_guid = guid;
}
public String getGuid() {
return m_guid;
}
public String getClientID() {
return m_clientID;
}
public void setClientID(String m_clientID) {
this.m_clientID = m_clientID;
}
public void setMessageID(Integer messageID) {
this.m_msgID = messageID;
}
public Integer getMessageID() {
return m_msgID;
}
public ByteBuffer getMessage() {
return ByteBuffer.wrap(m_payload);
}
public void setRetained(boolean retained) {
this.m_retained = retained;
}
public boolean isRetained() {
return m_retained;
}
@Override
public String toString() {
return "PublishEvent{" +
"m_msgID=" + m_msgID +
", clientID='" + m_clientID + '\'' +
", m_retain=" + m_retained +
", m_qos=" + m_qos +
", m_topic='" + m_topic + '\'' +
'}';
}
}
/**
@@ -61,7 +112,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, ByteBuffer message, AbstractMessage.QOSType qos);
void storeRetained(String topic, String guid);
/**
* Return a list of retained messages that satisfy the condition.
@@ -72,18 +123,18 @@ public interface IMessagesStore {
* Persist the message.
* @return the unique id in the storage (guid).
* */
String storePublishForFuture(PublishEvent evt);
String 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<PublishEvent> listMessagesInSession(Collection<String> guids);
List<StoredMessage> listMessagesInSession(Collection<String> guids);
void dropMessagesInSession(String clientID);
PublishEvent getMessageByGuid(String guid);
StoredMessage getMessageByGuid(String guid);
/**
* Return the next valid packetIdentifier for the given client session.
broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java
+43 -17
View File
@@ -23,8 +23,8 @@ import org.eclipse.moquette.server.netty.NettyChannel;
import org.eclipse.moquette.spi.ClientSession;
import org.eclipse.moquette.spi.IMatchingCondition;
import org.eclipse.moquette.spi.IMessagesStore;
import org.eclipse.moquette.spi.IMessagesStore.*;
import org.eclipse.moquette.spi.ISessionsStore;
import org.eclipse.moquette.spi.impl.events.*;
import org.eclipse.moquette.spi.impl.security.IAuthorizator;
import org.eclipse.moquette.spi.impl.subscriptions.Subscription;
import org.eclipse.moquette.spi.impl.subscriptions.SubscriptionsStore;
@@ -244,14 +244,14 @@ public class ProtocolProcessor {
* */
private void republishStoredInSession(ClientSession clientSession) {
LOG.trace("republishStoredInSession for client <{}>", clientSession);
List<PublishEvent> publishedEvents = clientSession.storedMessages();
List<StoredMessage> publishedEvents = clientSession.storedMessages();
if (publishedEvents.isEmpty()) {
LOG.info("No stored messages for client <{}>", clientSession.clientID);
return;
}
LOG.info("republishing stored messages to client <{}>", clientSession.clientID);
for (PublishEvent pubEvt : publishedEvents) {
for (StoredMessage pubEvt : publishedEvents) {
//TODO put in flight zone
directSend(clientSession, pubEvt.getTopic(), pubEvt.getQos(),
pubEvt.getMessage(), false, pubEvt.getMessageID());
@@ -266,6 +266,19 @@ public class ProtocolProcessor {
ClientSession targetSession = m_sessionsStore.sessionForClient(clientID);
targetSession.inFlightAcknowledged(messageID);
}
private static StoredMessage asStoredMessage(PublishMessage msg) {
StoredMessage stored = new StoredMessage(msg.getPayload().array(), msg.getQos(), msg.getTopicName());
stored.setRetained(msg.isRetainFlag());
stored.setMessageID(msg.getMessageID());
return stored;
}
private static StoredMessage asStoredMessage(WillMessage will) {
StoredMessage pub = new StoredMessage(will.getPayload().array(), will.getQos(), will.getTopic());
pub.setRetained(will.isRetained());
return pub;
}
public void processPublish(ServerChannel session, PublishMessage msg) {
LOG.trace("PUB --PUBLISH--> SRV executePublish invoked with {}", msg);
@@ -278,20 +291,21 @@ public class ProtocolProcessor {
return;
}
final AbstractMessage.QOSType qos = msg.getQos();
final ByteBuffer message = msg.getPayload();
boolean retain = msg.isRetainFlag();
final Integer messageID = msg.getMessageID();
LOG.info("PUBLISH from clientID <{}> on topic <{}> with QoS {}", clientID, topic, qos);
PublishEvent publishEvt = new PublishEvent(clientID, msg);
String guid = null;
StoredMessage toStoreMsg = asStoredMessage(msg);
toStoreMsg.setClientID(clientID);
if (qos == AbstractMessage.QOSType.MOST_ONE) { //QoS0
forward2Subscribers(publishEvt);
forward2Subscribers(toStoreMsg);
} else if (qos == AbstractMessage.QOSType.LEAST_ONE) { //QoS1
forward2Subscribers(publishEvt);
forward2Subscribers(toStoreMsg);
sendPubAck(clientID, messageID);
LOG.debug("replying with PubAck to MSG ID {}", messageID);
} else if (qos == AbstractMessage.QOSType.EXACTLY_ONCE) { //QoS2
m_messagesStore.storePublishForFuture(publishEvt);
guid = m_messagesStore.storePublishForFuture(toStoreMsg);
sendPubRec(clientID, messageID);
//Next the client will send us a pub rel
//NB publish to subscribers for QoS 2 happen upon PUBREL from publisher
@@ -302,7 +316,15 @@ public class ProtocolProcessor {
//QoS == 0 && retain => clean old retained
m_messagesStore.cleanRetained(topic);
} else {
m_messagesStore.storeRetained(topic, message, qos);
if (!msg.getPayload().hasRemaining()) {
m_messagesStore.cleanRetained(topic);
} else {
if (guid == null) {
//before wasn't stored
guid = m_messagesStore.storePublishForFuture(toStoreMsg);
}
m_messagesStore.storeRetained(topic, guid);
}
}
}
m_interceptor.notifyTopicPublished(msg, clientID);
@@ -319,9 +341,10 @@ public class ProtocolProcessor {
messageId = m_messagesStore.nextPacketID(clientID);
}
PublishEvent pub = new PublishEvent(will.getTopic(), will.getQos(), will.getPayload(), will.isRetained(),
clientID, messageId);
forward2Subscribers(pub);
StoredMessage tobeStored = asStoredMessage(will);
tobeStored.setClientID(clientID);
tobeStored.setMessageID(messageId);
forward2Subscribers(tobeStored);
}
@@ -329,7 +352,7 @@ public class ProtocolProcessor {
* Flood the subscribers with the message to notify. MessageID is optional and should only used for QoS 1 and 2
* */
//TODO rename to route2Subscribers
void forward2Subscribers(PublishEvent pubEvt) {
void forward2Subscribers(StoredMessage pubEvt) {
final String topic = pubEvt.getTopic();
final AbstractMessage.QOSType publishingQos = pubEvt.getQos();
final ByteBuffer origMessage = pubEvt.getMessage();
@@ -452,13 +475,16 @@ public class ProtocolProcessor {
int messageID = msg.getMessageID();
LOG.debug("PUB --PUBREL--> SRV processPubRel invoked for clientID {} ad messageID {}", clientID, messageID);
ClientSession targetSession = m_sessionsStore.sessionForClient(clientID);
PublishEvent evt = targetSession.storedMessage(messageID);
StoredMessage evt = targetSession.storedMessage(messageID);
forward2Subscribers(evt);
if (evt.isRetain()) {
if (evt.isRetained()) {
final String topic = evt.getTopic();
final AbstractMessage.QOSType qos = evt.getQos();
m_messagesStore.storeRetained(topic, evt.getMessage(), qos);
if (!evt.getMessage().hasRemaining()) {
m_messagesStore.cleanRetained(topic);
} else {
m_messagesStore.storeRetained(topic, evt.getGuid());
}
}
sendPubComp(clientID, messageID);
broker/src/main/java/org/eclipse/moquette/spi/impl/events/PublishEvent.java
-103
View File
@@ -1,103 +0,0 @@
/*
* 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 org.eclipse.moquette.spi.impl.events;
import org.eclipse.moquette.proto.messages.AbstractMessage.QOSType;
import org.eclipse.moquette.proto.messages.PublishMessage;
import java.nio.ByteBuffer;
/**
*
* @author andrea
*/
public class PublishEvent {
String m_topic;
QOSType m_qos;
//byte[] m_message;
ByteBuffer m_message;
boolean m_retain;
String m_clientID;
//Optional attribute, available only fo QoS 1 and 2
Integer m_msgID;
private String guid;
public PublishEvent(String topic, QOSType qos, ByteBuffer message, boolean retain,
String clientID, Integer msgID) {
m_topic = topic;
m_qos = qos;
m_message = message;
m_retain = retain;
m_clientID = clientID;
if (qos != QOSType.MOST_ONE) {
m_msgID = msgID;
}
}
public PublishEvent(String clientID, PublishMessage msg) {
m_clientID = clientID;
m_topic = msg.getTopicName();
m_qos = msg.getQos();
m_message = msg.getPayload();
m_retain = msg.isRetainFlag();
if (msg.getQos() != QOSType.MOST_ONE) {
m_msgID = msg.getMessageID();
}
}
public String getTopic() {
return m_topic;
}
public QOSType getQos() {
return m_qos;
}
public ByteBuffer getMessage() {
return m_message;
}
public boolean isRetain() {
return m_retain;
}
public String getClientID() {
return m_clientID;
}
public Integer getMessageID() {
return m_msgID;
}
public void setGuid(String guid) {
this.guid = guid;
}
public String getGuid() {
return guid;
}
@Override
public String toString() {
return "PublishEvent{" +
"m_msgID=" + m_msgID +
", clientID='" + m_clientID + '\'' +
", m_retain=" + m_retain +
", m_qos=" + m_qos +
", m_topic='" + m_topic + '\'' +
'}';
}
}
broker/src/main/java/org/eclipse/moquette/spi/impl/storage/StoredPublishEvent.java
-79
View File
@@ -1,79 +0,0 @@
/*
* 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 org.eclipse.moquette.spi.impl.storage;
import java.io.Serializable;
import java.nio.ByteBuffer;
import org.eclipse.moquette.spi.impl.events.PublishEvent;
import org.eclipse.moquette.proto.messages.AbstractMessage.QOSType;
/**
* Publish event serialized to the DB.
*
* @author andrea
*/
public class StoredPublishEvent implements Serializable {
String m_topic;
QOSType m_qos;
byte[] m_message;
boolean m_retain;
String m_clientID;
//Optional attribute, available only fo QoS 1 and 2
Integer m_msgID;
String m_guid;
public StoredPublishEvent(PublishEvent wrapped) {
m_topic = wrapped.getTopic();
m_qos = wrapped.getQos();
m_retain = wrapped.isRetain();
m_clientID = wrapped.getClientID();
m_msgID = wrapped.getMessageID();
ByteBuffer buffer = wrapped.getMessage();
m_message = new byte[buffer.remaining()];
buffer.get(m_message);
buffer.rewind();
m_guid = wrapped.getGuid();
}
public String getTopic() {
return m_topic;
}
public QOSType getQos() {
return m_qos;
}
public byte[] getMessage() {
return m_message;
}
public boolean isRetain() {
return m_retain;
}
public String getClientID() {
return m_clientID;
}
public Integer getMessageID() {
return m_msgID;
}
public String getGuid() {
return m_guid;
}
}
broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java
+26 -43
View File
@@ -17,13 +17,10 @@
package org.eclipse.moquette.spi.persistence;
import org.eclipse.moquette.proto.MQTTException;
import org.eclipse.moquette.proto.messages.AbstractMessage;
import org.eclipse.moquette.spi.ClientSession;
import org.eclipse.moquette.spi.IMatchingCondition;
import org.eclipse.moquette.spi.IMessagesStore;
import org.eclipse.moquette.spi.ISessionsStore;
import org.eclipse.moquette.spi.impl.events.PublishEvent;
import org.eclipse.moquette.spi.impl.storage.StoredPublishEvent;
import org.eclipse.moquette.spi.impl.subscriptions.Subscription;
import org.mapdb.DB;
import org.mapdb.DBMaker;
@@ -33,7 +30,6 @@ import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
import java.io.Serializable;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Executors;
@@ -63,9 +59,10 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore {
private static final Logger LOG = LoggerFactory.getLogger(MapDBPersistentStore.class);
private ConcurrentMap<String, StoredMessage> m_retainedStore;
//maps clientID -> guid
private ConcurrentMap<String, String> m_retainedStore;
//maps guid to message, it's message store
private ConcurrentMap<String, StoredPublishEvent> m_persistentMessageStore;
private ConcurrentMap<String, StoredMessage> m_persistentMessageStore;
//maps clientID->[MessageId -> guid]
private ConcurrentMap<String, Map<Integer, String>> m_inflightStore;
//map clientID <-> set of currently in flight packet identifiers
@@ -87,7 +84,6 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore {
/*
* The default constructor will create an in memory store as no file path was specified
*/
public MapDBPersistentStore() {
this.m_storePath = "";
}
@@ -134,16 +130,8 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore {
}
@Override
public void storeRetained(String topic, ByteBuffer message, AbstractMessage.QOSType qos) {
if (!message.hasRemaining()) {
//clean the message from topic
m_retainedStore.remove(topic);
} else {
//store the message to the topic
byte[] raw = new byte[message.remaining()];
message.get(raw);
m_retainedStore.put(topic, new StoredMessage(raw, qos, topic));
}
public void storeRetained(String topic, String guid) {
m_retainedStore.put(topic, guid);
}
@Override
@@ -151,8 +139,9 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore {
LOG.debug("searchMatching scanning all retained messages, presents are {}", m_retainedStore.size());
List<StoredMessage> results = new ArrayList<>();
for (Map.Entry<String, StoredMessage> entry : m_retainedStore.entrySet()) {
StoredMessage storedMsg = entry.getValue();
for (Map.Entry<String, String> entry : m_retainedStore.entrySet()) {
final String guid = entry.getValue();
StoredMessage storedMsg = m_persistentMessageStore.get(guid);
if (condition.match(entry.getKey())) {
results.add(storedMsg);
}
@@ -162,42 +151,50 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore {
}
@Override
public String storePublishForFuture(PublishEvent evt) {
public String 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();
evt.setGuid(guid);
m_persistentMessageStore.put(guid, convertToStored(evt));
ConcurrentMap<Integer, String> messageIdToGuid = m_db.getHashMap("guidsMapping" + evt.getClientID());
m_persistentMessageStore.put(guid, evt);
ConcurrentMap<Integer, String> messageIdToGuid = m_db.getHashMap(messageId2GuidsMapName(evt.getClientID()));
messageIdToGuid.put(evt.getMessageID(), guid);
return guid;
}
@Override
public List<PublishEvent> listMessagesInSession(Collection<String> guids) {
List<PublishEvent> ret = new ArrayList<>();
public List<StoredMessage> listMessagesInSession(Collection<String> guids) {
List<StoredMessage> ret = new ArrayList<>();
for (String guid : guids) {
ret.add(convertFromStored(m_persistentMessageStore.get(guid))) ;
ret.add(m_persistentMessageStore.get(guid));
}
return ret;
}
@Override
public void dropMessagesInSession(String clientID) {
m_db.getHashMap("guidsMapping" + clientID).clear();
m_db.getHashMap(messageId2GuidsMapName(clientID)).clear();
m_persistentMessageStore.remove(clientID);
}
@Override
public PublishEvent getMessageByGuid(String guid) {
return convertFromStored(m_persistentMessageStore.get(guid));
public StoredMessage getMessageByGuid(String guid) {
return m_persistentMessageStore.get(guid);
}
@Override
public String mapToGuid(String clientID, int messageID) {
ConcurrentMap<Integer, String> messageIdToGuid = m_db.getHashMap("guidsMapping" + clientID);
ConcurrentMap<Integer, String> messageIdToGuid = m_db.getHashMap(messageId2GuidsMapName(clientID));
return messageIdToGuid.get(messageID);
}
private String messageId2GuidsMapName(String clientID) {
return "guidsMapping_" + clientID;
}
/**
* Return the next valid packetIdentifier for the given client session.
* */
@@ -413,18 +410,4 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore {
this.m_scheduler.shutdown();
LOG.debug("Persistence commit scheduler is shutdown");
}
private StoredPublishEvent convertToStored(PublishEvent evt) {
return new StoredPublishEvent(evt);
}
private PublishEvent convertFromStored(StoredPublishEvent evt) {
byte[] message = evt.getMessage();
ByteBuffer bbmessage = ByteBuffer.wrap(message);
//bbmessage.flip();
PublishEvent pubEvt = new PublishEvent(evt.getTopic(), evt.getQos(),
bbmessage, evt.isRetain(), evt.getClientID(), evt.getMessageID());
pubEvt.setGuid(evt.getGuid());
return pubEvt;
}
}