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

Implemented packetID generation for QoS 1,2

Browse Source
This commit is contained in:
andrea
2015-03-10 22:52:08 +01:00
parent d271e23f99
commit 715937a394
6 changed files with 134 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/eclipse/moquette/server/netty/NettyChannel.java
+1 -1
View File
@@ -84,6 +84,6 @@ public class NettyChannel implements ServerChannel {
@Override
public String toString() {
String clientID = (String) getAttribute(Constants.ATTR_CLIENTID);
return "session [clientID "+ clientID +"]";
return "session [clientID: "+ clientID +"]" + super.toString();
}
}
broker/src/main/java/org/eclipse/moquette/spi/IMessagesStore.java
+8 -3
View File
@@ -77,13 +77,18 @@ public interface IMessagesStore {
*/
List<PublishEvent> listMessagesInSession(String clientID);
void removeMessageInSession(String clientID, int messageID);
void removeMessageInSession(String clientID, int packetID);
void dropMessagesInSession(String clientID);
void cleanInFlight(String msgID);
void cleanInFlight(String clientID, int packetID);
void addInFlight(PublishEvent evt, String publishKey);
void addInFlight(PublishEvent evt, String clientID, int packetID);
/**
* Return the next valid packetIdentifer for the given client session.
* */
int nextPacketID(String clientID);
void close();
broker/src/main/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java
+29 -3
View File
@@ -39,6 +39,7 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore {
//TODO move in a multimap because only Qos1 and QoS2 are stored here and they have messageID(key of secondary map)
private Map<String, List<PublishEvent>> m_persistentMessageStore = new HashMap<String, List<PublishEvent>>();
private Map<String, PublishEvent> m_inflightStore = new HashMap<String, PublishEvent>();
private Map<String, Set<Integer>> m_inflightIDs = new HashMap<>();
private Map<String, PublishEvent> m_qos2Store = new HashMap<String, PublishEvent>();
private static final Logger LOG = LoggerFactory.getLogger(MemoryStorageService.class);
@@ -119,15 +120,40 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore {
}
@Override
public void cleanInFlight(String msgID) {
m_inflightStore.remove(msgID);
public void cleanInFlight(String clientID, int packetID) {
String publishKey = String.format("%s%d", clientID, packetID);
m_inflightStore.remove(publishKey);
Set<Integer> inFlightForClient = m_inflightIDs.get(clientID);
if (inFlightForClient != null) {
inFlightForClient.remove(packetID);
}
}
@Override
public void addInFlight(PublishEvent evt, String publishKey) {
public void addInFlight(PublishEvent evt, String clientID, int packetID) {
String publishKey = String.format("%s%d", clientID, packetID);
m_inflightStore.put(publishKey, evt);
}
/**
* Return the next valid packetIdentifer for the given client session.
* */
@Override
public int nextPacketID(String clientID) {
Set<Integer> inFlightForClient = m_inflightIDs.get(clientID);
if (inFlightForClient == null) {
int nextPacketId = 1;
inFlightForClient = new HashSet<>();
inFlightForClient.add(nextPacketId);
m_inflightIDs.put(clientID, inFlightForClient);
return nextPacketId;
}
int maxId = Collections.max(inFlightForClient);
int nextPacketId = (maxId + 1) % 0xFFFF;
inFlightForClient.add(nextPacketId);
return nextPacketId;
}
@Override
public void addNewSubscription(Subscription newSubscription, String clientID) {
if (!m_persistentSubscriptions.containsKey(clientID)) {
broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java
+60 -63
View File
@@ -19,10 +19,7 @@ import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import java.nio.ByteBuffer;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import org.eclipse.moquette.spi.IMatchingCondition;
@@ -141,7 +138,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
@MQTTMessage(message = ConnectMessage.class)
void processConnect(ServerChannel session, ConnectMessage msg) {
LOG.debug("processConnect for client {}", msg.getClientID());
LOG.debug("CONNECT for client <{}>", msg.getClientID());
if (msg.getProcotolVersion() != VERSION_3_1 && msg.getProcotolVersion() != VERSION_3_1_1) {
ConnAckMessage badProto = new ConnAckMessage();
badProto.setReturnCode(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION);
@@ -182,6 +179,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
session.setAttribute(Constants.CLEAN_SESSION, msg.isCleanSession());
//used to track the client in the subscription and publishing phases.
session.setAttribute(Constants.ATTR_CLIENTID, msg.getClientID());
LOG.debug("Conent create session <{}>", session);
session.setIdleTime(Math.round(keepAlive * 1.5f));
@@ -223,13 +221,11 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
if (!msg.isCleanSession() && m_sessionsStore.contains(msg.getClientID())) {
okResp.setSessionPresent(true);
}
LOG.debug("processConnect sent OK ConnAck");
session.write(okResp);
LOG.info("Connected client ID <{}> with clean session {}", msg.getClientID(), msg.isCleanSession());
LOG.info("Create persistent session for clientID {}", msg.getClientID());
LOG.info("Create persistent session for clientID <{}>", msg.getClientID());
m_sessionsStore.addNewSubscription(Subscription.createEmptySubscription(msg.getClientID(), true), msg.getClientID()); //null means EmptySubscription
LOG.info("Connected client ID <{}> with clean session {}", msg.getClientID(), msg.isCleanSession());
if (!msg.isCleanSession()) {
//force the republish of stored QoS1 and QoS2
republishStoredInSession(msg.getClientID());
@@ -240,7 +236,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
* Republish QoS1 and QoS2 messages stored into the session for the clientID.
* */
private void republishStoredInSession(String clientID) {
LOG.trace("republishStoredInSession invoked");
LOG.trace("republishStoredInSession for client <{}>", clientID);
List<PublishEvent> publishedEvents = m_messagesStore.listMessagesInSession(clientID);
if (publishedEvents.isEmpty()) {
LOG.info("No stored messages for client <{}>", clientID);
@@ -285,37 +281,30 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
}
private void processPublish(String clientID, String topic, QOSType qos, ByteBuffer message, boolean retain, Integer messageID) {
LOG.info("Publish received from clientID <{}> on topic <{}> with QoS {}",
clientID, topic, qos);
LOG.info("PUBLISH from clientID <{}> on topic <{}> with QoS {}", clientID, topic, qos);
String publishKey = null;
if (qos == AbstractMessage.QOSType.LEAST_ONE) {
//store the temporary message
publishKey = String.format("%s%d", clientID, messageID);
if (qos == AbstractMessage.QOSType.MOST_ONE) { //QoS0
forward2Subscribers(topic, qos, message, retain, messageID);
} else if (qos == AbstractMessage.QOSType.LEAST_ONE) {
String publishKey = String.format("%s%d", clientID, messageID);
PublishEvent inFlightEvt = new PublishEvent(topic, qos, message, retain,
clientID, messageID);
m_messagesStore.addInFlight(inFlightEvt, publishKey);
//TODO use a message store for TO PUBLISH MESSAGES it has nothing to do with inFlight!!
m_messagesStore.addInFlight(inFlightEvt, clientID, messageID);
forward2Subscribers(topic, qos, message, retain, messageID);
m_messagesStore.cleanInFlight(clientID, messageID);
//NB the PUB_ACK could be sent also after the addInFlight
sendPubAck(new PubAckEvent(messageID, clientID));
LOG.debug("replying with PubAck to MSG ID {}", messageID);
} else if (qos == AbstractMessage.QOSType.EXACTLY_ONCE) {
publishKey = String.format("%s%d", clientID, messageID);
String publishKey = String.format("%s%d", clientID, messageID);
//store the message in temp store
PublishEvent qos2Persistent = new PublishEvent(topic, qos, message, retain,
clientID, messageID);
m_messagesStore.persistQoS2Message(publishKey, qos2Persistent);
sendPubRec(clientID, messageID);
}
//NB publish to subscribers for QoS 2 happen upon PUBREL from publisher
if (qos != AbstractMessage.QOSType.EXACTLY_ONCE) {
publish2Subscribers(topic, qos, message, retain, messageID);
}
if (qos == AbstractMessage.QOSType.LEAST_ONE) {
if (publishKey == null) {
throw new RuntimeException("Found a publish key null for QoS " + qos + " for message " + messageID);
}
m_messagesStore.cleanInFlight(publishKey);
sendPubAck(new PubAckEvent(messageID, clientID));
LOG.debug("replying with PubAck to MSG ID {}", messageID);
//Next the client will send us a pub rel
//NB publish to subscribers for QoS 2 happen upon PUBREL from publisher
}
if (retain) {
@@ -331,19 +320,23 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
/**
* Specialized version to publish will testament message.
*/
private void processPublish(WillMessage will, String clientID) {
private void forwardPublishWill(WillMessage will, String clientID) {
//TODO reimplement it!
//TODO it has just to publish the message downstream to the subscribers
final String topic = will.getTopic();
final AbstractMessage.QOSType qos = will.getQos();
final ByteBuffer message = will.getPayload();
boolean retain = will.isRetained();
processPublish(clientID, topic, qos, message, retain, null);
//NB it's a will publish, it needs a PacketIdentifier for this conn, default to 1
processPublish(clientID, topic, qos, message, retain, 1 /*null*/);
}
/**
* Flood the subscribers with the message to notify. MessageID is optional and should only used for QoS 1 and 2
* */
private void publish2Subscribers(String topic, AbstractMessage.QOSType qos, ByteBuffer origMessage, boolean retain, Integer messageID) {
LOG.debug("publish2Subscribers republishing to existing subscribers that matches the topic {}", topic);
private void forward2Subscribers(String topic, AbstractMessage.QOSType qos, ByteBuffer origMessage,
boolean retain, Integer messageID) {
LOG.debug("forward2Subscribers republishing to existing subscribers that matches the topic {}", topic);
if (LOG.isDebugEnabled()) {
LOG.debug("content <{}>", DebugUtils.payload2Str(origMessage));
LOG.debug("subscription tree {}", subscriptions.dumpTree());
@@ -353,13 +346,13 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
qos = sub.getRequestedQos();
}
ByteBuffer message = origMessage.duplicate();
LOG.debug("Broker republishing to client <{}> topic <{}> qos <{}>, active {}",
LOG.debug("Broker republishing to client <{}> topic <{}> qos <{}>, active {}",
sub.getClientId(), sub.getTopicFilter(), qos, sub.isActive());
ByteBuffer message = origMessage.duplicate();
if (qos == AbstractMessage.QOSType.MOST_ONE && sub.isActive()) {
//QoS 0
sendPublish(sub.getClientId(), topic, qos, message, false);
//forwardPublishQoS0(sub.getClientId(), topic, qos, message, false);
sendPublish(sub.getClientId(), topic, qos, message, false, null);
} else {
//QoS 1 or 2
//if the target subscription is not clean session and is not connected => store it
@@ -368,28 +361,23 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
PublishEvent newPublishEvt = new PublishEvent(topic, qos, message, retain, sub.getClientId(), messageID != null ? messageID : 0);
m_messagesStore.storePublishForFuture(newPublishEvt);
} else {
//TODO also QoS 1 has to be stored in Flight Zone
//if QoS 2 then store it in temp memory
if (qos == AbstractMessage.QOSType.EXACTLY_ONCE) {
String publishKey = String.format("%s%d", sub.getClientId(), messageID);
PublishEvent newPublishEvt = new PublishEvent(topic, qos, message, retain, sub.getClientId(), messageID != null ? messageID : 0);
m_messagesStore.addInFlight(newPublishEvt, publishKey);
m_messagesStore.addInFlight(newPublishEvt, sub.getClientId(), messageID);
}
//publish
if (sub.isActive()) {
sendPublish(sub.getClientId(), topic, qos, message, false);
int messageId = m_messagesStore.nextPacketID(sub.getClientId());
sendPublish(sub.getClientId(), topic, qos, message, false, messageId);
}
}
}
}
}
private void sendPublish(String clientId, String topic, AbstractMessage.QOSType qos, ByteBuffer message, boolean retained) {
//TODO pay attention to the message ID can't be 0 and it's the message sent to subscriber
int messageID = 1;
sendPublish(clientId, topic, qos, message, retained, messageID);
}
private void sendPublish(String clientId, String topic, AbstractMessage.QOSType qos, ByteBuffer message, boolean retained, int messageID) {
private void sendPublish(String clientId, String topic, AbstractMessage.QOSType qos, ByteBuffer message, boolean retained, Integer messageID) {
LOG.debug("sendPublish invoked clientId <{}> on topic <{}> QoS {} retained {} messageID {}", clientId, topic, qos, retained, messageID);
PublishMessage pubMessage = new PublishMessage();
pubMessage.setRetainFlag(retained);
@@ -401,8 +389,13 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
if (LOG.isDebugEnabled()) {
LOG.debug("content <{}>", DebugUtils.payload2Str(message));
}
//set the PacketIdentifier only for QoS > 0
if (pubMessage.getQos() != AbstractMessage.QOSType.MOST_ONE) {
pubMessage.setMessageID(messageID);
} else {
if (messageID != null) {
throw new RuntimeException("Internal bad error, trying to forwardPublish a QoS 0 message with PacketIdentifier: " + messageID);
}
}
if (m_clientIDs == null) {
@@ -413,7 +406,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client <%s> in cache <%s>", clientId, m_clientIDs));
}
LOG.debug("Session for clientId {} is {}", clientId, m_clientIDs.get(clientId).getSession());
// m_clientIDs.get(clientId).getSession().write(pubMessage);
// m_clientIDs.get(clientId).getSession().write(pubMessage);
disruptorPublish(new OutputMessagingEvent(m_clientIDs.get(clientId).getSession(), pubMessage));
}
@@ -465,7 +458,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
final String topic = evt.getTopic();
final AbstractMessage.QOSType qos = evt.getQos();
publish2Subscribers(topic, qos, evt.getMessage(), evt.isRetain(), evt.getMessageID());
forward2Subscribers(topic, qos, evt.getMessage(), evt.isRetain(), evt.getMessageID());
m_messagesStore.removeQoS2Message(publishKey);
@@ -496,7 +489,8 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
pubRelMessage.setQos(AbstractMessage.QOSType.LEAST_ONE);
// m_clientIDs.get(clientID).getSession().write(pubRelMessage);
disruptorPublish(new OutputMessagingEvent(m_clientIDs.get(clientID).getSession(), pubRelMessage));
//disruptorPublish(new OutputMessagingEvent(m_clientIDs.get(clientID).getSession(), pubRelMessage));
session.write(pubRelMessage);
}
@MQTTMessage(message = PubCompMessage.class)
@@ -505,8 +499,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
int messageID = msg.getMessageID();
LOG.debug("\t\tSRV <--PUBCOMP-- SUB processPubComp invoked for clientID {} ad messageID {}", clientID, messageID);
//once received the PUBCOMP then remove the message from the temp memory
String publishKey = String.format("%s%d", clientID, messageID);
m_messagesStore.cleanInFlight(publishKey);
m_messagesStore.cleanInFlight(clientID, messageID);
}
@MQTTMessage(message = DisconnectMessage.class)
@@ -526,7 +519,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
//cleanup the will store
m_willStore.remove(clientID);
LOG.info("Disconnected client <{}> with clean session {}", clientID, cleanSession);
LOG.info("DISCONNECT client <{}> with clean session {}", clientID, cleanSession);
}
void processConnectionLost(LostConnectionEvent evt) {
@@ -548,7 +541,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
//publish the Will message (if any) for the clientID
if (m_willStore.containsKey(clientID)) {
WillMessage will = m_willStore.get(clientID);
processPublish(will, clientID);
forwardPublishWill(will, clientID);
m_willStore.remove(clientID);
}
}
@@ -562,7 +555,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
List<String> topics = msg.topicFilters();
int messageID = msg.getMessageID();
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
LOG.debug("processUnsubscribe invoked, removing subscription on topics {}, for clientID <{}>", topics, clientID);
LOG.debug("UNSUBSCRIBE subscription on topics {} for clientID <{}>", topics, clientID);
for (String topic : topics) {
subscriptions.removeSubscription(topic, clientID);
@@ -579,7 +572,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
void processSubscribe(ServerChannel session, SubscribeMessage msg) {
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
boolean cleanSession = (Boolean) session.getAttribute(Constants.CLEAN_SESSION);
LOG.debug("processSubscribe invoked from client {} with msgID {}", clientID, msg.getMessageID());
LOG.debug("SUBSCRIBE client <{}> packetID {}", clientID, msg.getMessageID());
for (SubscribeMessage.Couple req : msg.subscriptions()) {
AbstractMessage.QOSType qos = AbstractMessage.QOSType.values()[req.getQos()];
@@ -597,7 +590,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
ackMessage.addType(qos);
}
LOG.debug("replying with SubAck to MSG ID {}", msg.getMessageID());
LOG.debug("SUBACK for packetID {}", msg.getMessageID());
session.write(ackMessage);
}
@@ -619,7 +612,10 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
for (IMessagesStore.StoredMessage storedMsg : messages) {
//fire the as retained the message
LOG.debug("send publish message for topic {}", topic);
sendPublish(newSubscription.getClientId(), storedMsg.getTopic(), storedMsg.getQos(), storedMsg.getPayload(), true);
//forwardPublishQoS0(newSubscription.getClientId(), storedMsg.getTopic(), storedMsg.getQos(), storedMsg.getPayload(), true);
Integer packetID = storedMsg.getQos() == QOSType.MOST_ONE ? null :
m_messagesStore.nextPacketID(newSubscription.getClientId());
sendPublish(newSubscription.getClientId(), storedMsg.getTopic(), storedMsg.getQos(), storedMsg.getPayload(), true, packetID);
}
}
@@ -637,6 +633,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
MessagingEvent evt = t.getEvent();
//It's always of type OutputMessagingEvent
OutputMessagingEvent outEvent = (OutputMessagingEvent) evt;
LOG.debug("Output event, sending {}", outEvent.getMessage());
outEvent.getChannel().write(outEvent.getMessage());
}
broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java
+36 -6
View File
@@ -50,6 +50,8 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore {
private ConcurrentMap<String, List<StoredPublishEvent>> m_persistentMessageStore;
//bind clientID+MsgID -> evt message published
private ConcurrentMap<String, StoredPublishEvent> m_inflightStore;
//map clientID <-> set of currently in flight packet identifiers
Map<String, Set<Integer>> m_inFlightIds;
//bind clientID+MsgID -> evt message published
private ConcurrentMap<String, StoredPublishEvent> m_qos2Store;
//persistent Map of clientID, set of Subscriptions
@@ -66,11 +68,11 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore {
LOG.error(null, ex);
throw new MQTTException("Can't create temp file for subscriptions storage [" + STORAGE_FILE_PATH + "]", ex);
}
m_db = DBMaker.newFileDB(tmpFile)
.make();
m_db = DBMaker.newFileDB(tmpFile).make();
m_retainedStore = m_db.getHashMap("retained");
m_persistentMessageStore = m_db.getHashMap("persistedMessages");
m_inflightStore = m_db.getHashMap("inflight");
m_inFlightIds = m_db.getHashMap("inflightPacketIDs");
m_persistentSubscriptions = m_db.getHashMap("subscriptions");
m_qos2Store = m_db.getHashMap("qos2Store");
}
@@ -115,7 +117,7 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore {
List<StoredPublishEvent> storedEvents;
String clientID = evt.getClientID();
if (!m_persistentMessageStore.containsKey(clientID)) {
storedEvents = new ArrayList<StoredPublishEvent>();
storedEvents = new ArrayList<>();
} else {
storedEvents = m_persistentMessageStore.get(clientID);
}
@@ -159,17 +161,45 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore {
m_db.commit();
}
public void cleanInFlight(String msgID) {
m_inflightStore.remove(msgID);
//----------------- In flight methods -----------------
@Override
public void cleanInFlight(String clientID, int packetID) {
String publishKey = String.format("%s%d", clientID, packetID);
m_inflightStore.remove(publishKey);
Set<Integer> inFlightForClient = this.m_inFlightIds.get(clientID);
if (inFlightForClient != null) {
inFlightForClient.remove(packetID);
}
m_db.commit();
}
public void addInFlight(PublishEvent evt, String publishKey) {
@Override
public void addInFlight(PublishEvent evt, String clientID, int packetID) {
String publishKey = String.format("%s%d", clientID, packetID);
StoredPublishEvent storedEvt = convertToStored(evt);
m_inflightStore.put(publishKey, storedEvt);
m_db.commit();
}
/**
* Return the next valid packetIdentifer for the given client session.
* */
@Override
public int nextPacketID(String clientID) {
Set<Integer> inFlightForClient = this.m_inFlightIds.get(clientID);
if (inFlightForClient == null) {
int nextPacketId = 1;
inFlightForClient = new HashSet<>();
inFlightForClient.add(nextPacketId);
this.m_inFlightIds.put(clientID, inFlightForClient);
return nextPacketId;
}
int maxId = Collections.max(inFlightForClient);
int nextPacketId = (maxId + 1) % 0xFFFF;
inFlightForClient.add(nextPacketId);
return nextPacketId;
}
public void addNewSubscription(Subscription newSubscription, String clientID) {
LOG.debug("addNewSubscription invoked with subscription {} for client {}", newSubscription, clientID);
if (!m_persistentSubscriptions.containsKey(clientID)) {
broker/src/test/java/org/eclipse/moquette/server/ServerIntegrationPahoTest.java
-1
View File
@@ -173,7 +173,6 @@ public class ServerIntegrationPahoTest {
LOG.info("*** testRetain_maintainMessage_againstClientDestruction ***");
m_client.connect();
m_client.publish("/topic", "Test my payload".getBytes(), 1, true);
m_client.disconnect();
//reconnect and publish