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

Added management of PUBACK messages in return of re-publish messages (after a cleanSession = false rebooted client) to clean sequentially the already dispatched messages

Browse Source
This commit is contained in:
andrea
2013-11-27 18:47:34 +01:00
parent 6c3deb392d
commit 4226f4134f
6 changed files with 47 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java
+2
View File
@@ -26,6 +26,8 @@ public interface IStorageService {
void storePublishForFuture(PublishEvent evt);
List<PublishEvent> retrivePersistedPublishes(String clientID);
void cleanPersistedPublishMessage(String clientID, int messageID);
void cleanPersistedPublishes(String clientID);
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java
+16
View File
@@ -55,6 +55,7 @@ public class HawtDBStorageService implements IStorageService {
private PageFileFactory pageFactory;
//maps clientID to the list of pending messages stored
//TODO move in a multimap because only Qos1 and QoS2 are stored here and they have messageID(key of secondary map)
private SortedIndex<String, List<StoredPublishEvent>> m_persistentMessageStore;
private SortedIndex<String, StoredMessage> m_retainedStore;
//bind clientID+MsgID -> evt message published
@@ -185,6 +186,21 @@ public class HawtDBStorageService implements IStorageService {
}
return liveEvts;
}
public void cleanPersistedPublishMessage(String clientID, int messageID) {
List<StoredPublishEvent> events = m_persistentMessageStore.get(clientID);
if (events == null) {
return;
}
StoredPublishEvent toRemoveEvt = null;
for (StoredPublishEvent evt : events) {
if (evt.getMessageID() == messageID) {
toRemoveEvt = evt;
}
}
events.remove(toRemoveEvt);
m_persistentMessageStore.put(clientID, events);
}
public void cleanPersistedPublishes(String clientID) {
m_persistentMessageStore.remove(clientID);
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java
+13
View File
@@ -23,6 +23,7 @@ public class MemoryStorageService implements IStorageService {
private Map<String, Set<Subscription>> m_persistentSubscriptions = new HashMap<String, Set<Subscription>>();
private Map<String, HawtDBStorageService.StoredMessage> m_retainedStore = new HashMap<String, HawtDBStorageService.StoredMessage>();
//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, PublishEvent> m_qos2Store = new HashMap<String, PublishEvent>();
@@ -76,6 +77,18 @@ public class MemoryStorageService implements IStorageService {
public List<PublishEvent> retrivePersistedPublishes(String clientID) {
return m_persistentMessageStore.get(clientID);
}
public void cleanPersistedPublishMessage(String clientID, int messageID) {
List<PublishEvent> events = m_persistentMessageStore.get(clientID);
PublishEvent toRemoveEvt = null;
for (PublishEvent evt : events) {
if (evt.getMessageID() == messageID) {
toRemoveEvt = evt;
}
}
events.remove(toRemoveEvt);
m_persistentMessageStore.put(clientID, events);
}
public void cleanPersistedPublishes(String clientID) {
m_persistentMessageStore.remove(clientID);
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java
+10 -5
View File
@@ -167,7 +167,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
}
private void republishStored(String clientID) {
LOG.debug("republishStored invoked");
LOG.trace("republishStored invoked");
List<PublishEvent> publishedEvents = m_storageService.retrivePersistedPublishes(clientID);
if (publishedEvents == null) {
LOG.info("No stored messages for client <{}>", clientID);
@@ -181,6 +181,11 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
}
}
void processPubAck(String clientID, int messageID) {
//Remove the message from message store
m_storageService.cleanPersistedPublishMessage(clientID, messageID);
}
private void processRemoveAllSubscriptions(String clientID) {
LOG.info("cleaning old saved subscriptions for client <{}>", clientID);
subscriptions.removeForClient(clientID);
@@ -190,7 +195,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
}
protected void processPublish(PublishEvent evt) {
LOG.debug("processPublish invoked with " + evt);
LOG.trace("processPublish invoked with {}", evt);
final String topic = evt.getTopic();
final AbstractMessage.QOSType qos = evt.getQos();
final ByteBuffer message = evt.getMessage();
@@ -287,13 +292,13 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
LOG.debug("Session for clientId {} is {}", clientId, m_clientIDs.get(clientId).getSession());
// m_clientIDs.get(clientId).getSession().write(pubMessage);
disruptorPublish(new OutputMessagingEvent(m_clientIDs.get(clientId).getSession(), pubMessage));
}catch(Throwable t) {
} catch(Throwable t) {
LOG.error(null, t);
}
}
private void sendPubRec(String clientID, int messageID) {
LOG.debug("sendPubRec invoked for clientID {} with messageID {}", clientID, messageID);
LOG.trace("sendPubRec invoked for clientID {} with messageID {}", clientID, messageID);
PubRecMessage pubRecMessage = new PubRecMessage();
pubRecMessage.setMessageID(messageID);
@@ -302,7 +307,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
}
private void sendPubAck(PubAckEvent evt) {
LOG.debug("sendPubAck invoked");
LOG.trace("sendPubAck invoked");
String clientId = evt.getClientID();
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
+4
View File
@@ -148,6 +148,10 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
int messageID = ((PubCompMessage) message).getMessageID();
m_processor.processPubComp(clientID, messageID);
} else if (message instanceof PubAckMessage) {
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
int messageID = ((PubAckMessage) message).getMessageID();
m_processor.processPubAck(clientID, messageID);
} else {
throw new RuntimeException("Illegal message received " + message);
}
broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyMQTTHandler.java
+2 -1
View File
@@ -23,7 +23,7 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter {
private static final Logger LOG = LoggerFactory.getLogger(NettyMQTTHandler.class);
private IMessaging m_messaging;
private Map<ChannelHandlerContext, NettyChannel> m_channelMapper = new HashMap<ChannelHandlerContext, NettyChannel>();
private final Map<ChannelHandlerContext, NettyChannel> m_channelMapper = new HashMap<ChannelHandlerContext, NettyChannel>();
@Override
public void channelRead(ChannelHandlerContext ctx, Object message) {
@@ -39,6 +39,7 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter {
case PUBCOMP:
case PUBREL:
case DISCONNECT:
case PUBACK:
NettyChannel channel;
synchronized(m_channelMapper) {
if (!m_channelMapper.containsKey(ctx)) {