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

Refactored the part of move from inFlight to second stage rel ack, this driven to missed notification to the interceptors

Browse Source
This commit is contained in:
andsel
2016-06-16 23:01:48 +02:00
parent d5399995c3
commit 0c4f7fa393
8 changed files with 106 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java
+1
View File
@@ -82,6 +82,7 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter {
}
} catch (Exception ex) {
LOG.error("Bad error in processing the message", ex);
ctx.fireExceptionCaught(ex);
}
}
broker/src/main/java/io/moquette/spi/ClientSession.java
+13 -9
View File
@@ -72,9 +72,6 @@ public class ClientSession {
this.m_sessionsStore = sessionsStore;
this.cleanSession = cleanSession;
}
//List of client's subscriptions
//list of messages not acknowledged by client
//list of in-flight messages
/**
* @return the list of messages to be delivered for client related to the session.
@@ -174,19 +171,18 @@ public class ClientSession {
}
public void inFlightAcknowledged(int messageID) {
LOG.trace("Acknowledging inflight, clientID <{}> messageID {}", this.clientID, messageID);
m_sessionsStore.inFlightAck(this.clientID, messageID);
}
public void inFlightAckWaiting(String guid, int messageID) {
LOG.trace("Adding to inflight {}, guid <{}>", messageID, guid);
m_sessionsStore.inFlight(this.clientID, messageID, guid);
}
public void secondPhaseAcknowledged(int messageID) {
m_sessionsStore.secondPhaseAcknowledged(clientID, messageID);
}
public void secondPhaseAckWaiting(int messageID) {
m_sessionsStore.secondPhaseAckWaiting(clientID, messageID);
public IMessagesStore.StoredMessage secondPhaseAcknowledged(int messageID) {
String guid = m_sessionsStore.secondPhaseAcknowledged(clientID, messageID);
return messagesStore.getMessageByGuid(guid);
}
public void enqueueToDeliver(String guid) {
@@ -209,4 +205,12 @@ public class ClientSession {
public AbstractMessage dequeue() {
return m_queueToPublish.poll();
}
public void moveInFlightToSecondPhaseAckWaiting(int messageID) {
m_sessionsStore.moveInFlightToSecondPhaseAckWaiting(this.clientID, messageID);
}
public IMessagesStore.StoredMessage getInflightMessage(int messageID) {
return m_sessionsStore.getInflightMessage(clientID, messageID);
}
}
broker/src/main/java/io/moquette/spi/ISessionsStore.java
+8 -5
View File
@@ -28,8 +28,6 @@ import io.moquette.spi.impl.subscriptions.Subscription;
*/
public interface ISessionsStore {
void updateCleanStatus(String clientID, boolean cleanSession);
class ClientTopicCouple {
public final String topicFilter;
public final String clientID;
@@ -69,6 +67,8 @@ public interface ISessionsStore {
void initStore();
void updateCleanStatus(String clientID, boolean cleanSession);
/**
* Add a new subscription to the session
*/
@@ -134,11 +134,14 @@ public interface ISessionsStore {
* */
void removeEnqueued(String clientID, String guid);
void secondPhaseAcknowledged(String clientID, int messageID);
void moveInFlightToSecondPhaseAckWaiting(String clientID, int messageID);
void secondPhaseAckWaiting(String clientID, int messageID);
/**
* @return the guid of message just acked.
* */
String secondPhaseAcknowledged(String clientID, int messageID);
String mapToGuid(String clientID, int messageID);
public StoredMessage getInflightMessage(String clientID, int messageID);
StoredMessage getInflightMessage(String clientID, int messageID);
}
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+29 -20
View File
@@ -295,11 +295,12 @@ public class ProtocolProcessor {
String clientID = NettyUtils.clientID(channel);
int messageID = msg.getMessageID();
String username = NettyUtils.userName(channel);
StoredMessage inflightMsg = m_sessionsStore.getInflightMessage(clientID, messageID);
LOG.trace("retrieving inflight for messageID <{}>", messageID);
//Remove the message from message store
ClientSession targetSession = m_sessionsStore.sessionForClient(clientID);
verifyToActivate(clientID, targetSession);
StoredMessage inflightMsg = targetSession.getInflightMessage(messageID);
targetSession.inFlightAcknowledged(messageID);
String topic = inflightMsg.getTopic();
@@ -448,7 +449,9 @@ public class ProtocolProcessor {
guid = m_messagesStore.storePublishForFuture(pubMsg);
}
for (final Subscription sub : subscriptions.matches(topic)) {
List<Subscription> topicMatchingSubscriptions = subscriptions.matches(topic);
LOG.trace("Found {} matching subscriptions to <{}>", topicMatchingSubscriptions.size(), topic);
for (final Subscription sub : topicMatchingSubscriptions) {
AbstractMessage.QOSType qos = publishingQos;
if (qos.byteValue() > sub.getRequestedQos().byteValue()) {
qos = sub.getRequestedQos();
@@ -509,7 +512,6 @@ public class ProtocolProcessor {
throw new RuntimeException("Internal bad error, found m_clientIDs to null while it should be " +
"initialized, somewhere it's overwritten!!");
}
//LOG.trace("clientIDs are {}", m_clientIDs);
if (m_clientIDs.get(clientId) == null) {
//TODO while we were publishing to the target client, that client disconnected,
// could happen is not an error HANDLE IT
@@ -588,12 +590,11 @@ public class ProtocolProcessor {
public void processPubRec(Channel channel, PubRecMessage msg) {
String clientID = NettyUtils.clientID(channel);
int messageID = msg.getMessageID();
ClientSession targetSession = m_sessionsStore.sessionForClient(clientID);
verifyToActivate(clientID, targetSession);
//remove from the inflight and move to the QoS2 second phase queue
targetSession.inFlightAcknowledged(messageID);
targetSession.secondPhaseAckWaiting(messageID);
int messageID = msg.getMessageID();
targetSession.moveInFlightToSecondPhaseAckWaiting(messageID);
//once received a PUBREC reply with a PUBREL(messageID)
LOG.debug("\t\tSRV <--PUBREC-- SUB processPubRec invoked for clientID {} ad messageID {}", clientID, messageID);
PubRelMessage pubRelMessage = new PubRelMessage();
@@ -606,16 +607,14 @@ public class ProtocolProcessor {
public void processPubComp(Channel channel, PubCompMessage msg) {
String clientID = NettyUtils.clientID(channel);
int messageID = msg.getMessageID();
StoredMessage inflightMsg = m_sessionsStore.getInflightMessage( clientID, messageID );
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
ClientSession targetSession = m_sessionsStore.sessionForClient(clientID);
verifyToActivate(clientID, targetSession);
targetSession.secondPhaseAcknowledged(messageID);
StoredMessage inflightMsg = targetSession.secondPhaseAcknowledged(messageID);
String username = NettyUtils.userName(channel);
String topic = inflightMsg.getTopic();
m_interceptor.notifyMessageAcknowledged( new InterceptAcknowledgedMessage(inflightMsg, topic, username) );
m_interceptor.notifyMessageAcknowledged(new InterceptAcknowledgedMessage(inflightMsg, topic, username));
}
public void processDisconnect(Channel channel) throws InterruptedException {
@@ -725,16 +724,25 @@ public class ProtocolProcessor {
if (LOG.isTraceEnabled()) {
LOG.trace("subscription tree {}", subscriptions.dumpTree());
}
for (Subscription subscription : newSubscriptions) {
subscribeSingleTopic(subscription);
}
channel.writeAndFlush(ackMessage);
//fire the publish
for(Subscription subscription : newSubscriptions) {
subscribeSingleTopic(subscription, username);
//fire the persisted messages in session
for (Subscription subscription : newSubscriptions) {
publishStoredMessagesInSession(subscription, username);
}
}
private boolean subscribeSingleTopic(final Subscription newSubscription, String username) {
private void subscribeSingleTopic(final Subscription newSubscription) {
LOG.debug("Subscribing {}", newSubscription);
subscriptions.add(newSubscription.asClientTopicCouple());
}
private void publishStoredMessagesInSession(final Subscription newSubscription, String username) {
LOG.debug("Publish persisted messages in session {}", newSubscription);
//scans retained messages to be published to the new subscription
//TODO this is ugly, it does a linear scan on potential big dataset
@@ -748,17 +756,18 @@ public class ProtocolProcessor {
ClientSession targetSession = m_sessionsStore.sessionForClient(newSubscription.getClientId());
verifyToActivate(newSubscription.getClientId(), targetSession);
for (IMessagesStore.StoredMessage storedMsg : messages) {
//fire the as retained the message
LOG.debug("send publish message for topic {}", newSubscription.getTopicFilter());
//forwardPublishQoS0(newSubscription.getClientId(), storedMsg.getTopic(), storedMsg.getQos(), storedMsg.getPayload(), true);
Integer packetID = storedMsg.getQos() == QOSType.MOST_ONE ? null :
targetSession.nextPacketId();
//fire as retained the message
LOG.trace("send publish message for topic {}", newSubscription.getTopicFilter());
Integer packetID = storedMsg.getQos() == QOSType.MOST_ONE ? null : targetSession.nextPacketId();
if (packetID != null) {
LOG.trace("Adding to inflight <{}>", packetID);
targetSession.inFlightAckWaiting(storedMsg.getGuid(), packetID);
}
directSend(targetSession, storedMsg.getTopic(), storedMsg.getQos(), storedMsg.getPayload(), true, packetID);
}
//notify the Observables
m_interceptor.notifyTopicSubscribed(newSubscription, username);
return true;
}
public void notifyChannelWritable(Channel channel) {
broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java
+1
View File
@@ -81,6 +81,7 @@ class MapDBMessagesStore implements IMessagesStore {
}
String guid = 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()));
messageIdToGuid.put(evt.getMessageID(), guid);
broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java
+33 -14
View File
@@ -45,8 +45,8 @@ class MapDBSessionsStore implements ISessionsStore {
private ConcurrentMap<String, PersistentSession> m_persistentSessions;
//maps clientID->[guid*], insertion order cares, it's queue
private ConcurrentMap<String, List<String>> m_enqueuedStore;
//maps clientID->[messageID*]
private ConcurrentMap<String, Set<Integer>> m_secondPhaseStore;
//maps clientID->[MessageId -> guid]
private ConcurrentMap<String, Map<Integer, String>> m_secondPhaseStore;
private final DB m_db;
private final IMessagesStore m_messagesStore;
@@ -172,6 +172,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);
if (m == null) {
LOG.error("Can't find the inFlight record for client <{}>", clientID);
@@ -193,6 +194,7 @@ class MapDBSessionsStore implements ISessionsStore {
m = new HashMap<>();
}
m.put(messageID, guid);
LOG.info("storing inflight clientID <{}> messageID {} guid <{}>", clientID, messageID, guid);
this.m_inflightStore.put(clientID, m);
}
@@ -216,17 +218,33 @@ class MapDBSessionsStore implements ISessionsStore {
}
@Override
public void secondPhaseAcknowledged(String clientID, int messageID) {
Set<Integer> messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, new HashSet<Integer>());
messageIDs.remove(messageID);
public void moveInFlightToSecondPhaseAckWaiting(String clientID, int messageID) {
LOG.info("acknowledging inflight clientID <{}> messageID {}", clientID, messageID);
Map<Integer, String> 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);
//remove from the ids store
Set<Integer> inFlightForClient = this.m_inFlightIds.get(clientID);
if (inFlightForClient != null) {
inFlightForClient.remove(messageID);
}
LOG.info("Moving to second phase store");
Map<Integer, String> messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, new HashMap<Integer, String>());
messageIDs.put(messageID, guid);
m_secondPhaseStore.put(clientID, messageIDs);
}
@Override
public void secondPhaseAckWaiting(String clientID, int messageID) {
Set<Integer> messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, new HashSet<Integer>());
messageIDs.add(messageID);
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);
m_secondPhaseStore.put(clientID, messageIDs);
return guid;
}
@Override
@@ -240,13 +258,14 @@ class MapDBSessionsStore implements ISessionsStore {
}
@Override
public StoredMessage getInflightMessage( String clientID, int messageID ) {
Map<Integer, String> clientEntries = m_inflightStore.get( clientID );
if( clientEntries == null )
public StoredMessage getInflightMessage(String clientID, int messageID) {
Map<Integer, String> clientEntries = m_inflightStore.get(clientID);
if (clientEntries == null)
return null;
String guid = clientEntries.get( messageID );
if( guid == null )
String guid = clientEntries.get(messageID);
LOG.info("inflight messageID {} guid <{}>", messageID, guid);
if (guid == null)
return null;
return m_messagesStore.getMessageByGuid( guid );
return m_messagesStore.getMessageByGuid(guid);
}
}