From d4cc804624b616b581d7ca7049a4bdc8480ed134 Mon Sep 17 00:00:00 2001 From: andrea Date: Mon, 25 Nov 2013 18:51:35 +0100 Subject: [PATCH] Added handling of dirty connection close from client, without an DISCONNECT message --- .../moquette/messaging/spi/IMessaging.java | 2 ++ .../messaging/spi/impl/ProtocolProcessor.java | 20 +++++++++++++++++-- .../messaging/spi/impl/SimpleMessaging.java | 7 +++++++ .../spi/impl/events/LostConnectionEvent.java | 19 ++++++++++++++++++ .../server/netty/NettyMQTTHandler.java | 4 ++++ 5 files changed, 50 insertions(+), 2 deletions(-) create mode 100644 broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/LostConnectionEvent.java diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java index 8fbd48a1..4a6001cb 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java @@ -13,6 +13,8 @@ public interface IMessaging { void stop(); void disconnect(ServerChannel session); + + void lostConnection(String clientID); void handleProtocolMessage(ServerChannel session, AbstractMessage msg); } diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java index 1cda6b42..04028e3b 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java @@ -192,6 +192,9 @@ class ProtocolProcessor implements EventHandler { final AbstractMessage.QOSType qos = evt.getQos(); final byte[] message = evt.getMessage(); boolean retain = evt.isRetain(); + + LOG.info(String.format("Publish recieved from clientID <%s> on topic <%s> with QoS %s, content [%s]", + evt.getClientID(), evt.getTopic(), evt.getQos(), new String(evt.getMessage()))); String publishKey = null; if (qos == AbstractMessage.QOSType.LEAST_ONE) { @@ -227,7 +230,8 @@ class ProtocolProcessor implements EventHandler { private void publish2Subscribers(String topic, AbstractMessage.QOSType qos, byte[] message, boolean retain, Integer messageID) { LOG.debug("publish2Subscribers republishing to existing subscribers that matches the topic " + topic); for (final Subscription sub : subscriptions.matches(topic)) { - LOG.debug("found matching subscription on topic " + sub.getTopic()); + LOG.debug(String.format("Broker publishing to client <%s> topic <%s> qos <%s> content [%s]", + sub.getClientId(), sub.getTopic(), qos, new String(message))); if (qos == AbstractMessage.QOSType.MOST_ONE) { //QoS 0 sendPublish(sub.getClientId(), topic, qos, message, false); @@ -263,6 +267,8 @@ class ProtocolProcessor implements EventHandler { pubMessage.setTopicName(topic); pubMessage.setQos(qos); pubMessage.setPayload(message); + LOG.info(String.format("send publish message to <%s> on topic <%s>, message is [%s]", + clientId, topic, new String(message))); if (pubMessage.getQos() != AbstractMessage.QOSType.MOST_ONE) { pubMessage.setMessageID(messageID); } @@ -377,7 +383,17 @@ class ProtocolProcessor implements EventHandler { //de-activate the subscriptions for this ClientID // String clientID = (String) evt.getSession().getAttribute(Constants.ATTR_CLIENTID); subscriptions.deactivate(clientID); - LOG.info(String.format("Disconnected client <%s>", clientID)); + LOG.info(String.format("Disconnected client <%s> with clean session %s", clientID, cleanSession)); + } + + void proccessConnectionLost(String clientID) { + //If already removed a disconnect message was already processed for this clientID + if (m_clientIDs.remove(clientID) != null) { + + //de-activate the subscriptions for this ClientID + subscriptions.deactivate(clientID); + LOG.info(String.format("Lost connection with client <%s>", clientID)); + } } /** diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java index a876d165..c8d6b608 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java @@ -82,6 +82,10 @@ public class SimpleMessaging implements IMessaging, EventHandler { public void disconnect(ServerChannel session) { disruptorPublish(new DisconnectEvent(session)); } + + public void lostConnection(String clientID) { + disruptorPublish(new LostConnectionEvent(clientID)); + } public void handleProtocolMessage(ServerChannel session, AbstractMessage msg) { disruptorPublish(new ProtocolEvent(session, msg)); @@ -153,6 +157,9 @@ public class SimpleMessaging implements IMessaging, EventHandler { } else if (evt instanceof InitEvent) { processInit(); + } else if (evt instanceof LostConnectionEvent) { + LostConnectionEvent lostEvt = (LostConnectionEvent) evt; + m_processor.proccessConnectionLost(lostEvt.getClientID()); } } diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/LostConnectionEvent.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/LostConnectionEvent.java new file mode 100644 index 00000000..58ff45b5 --- /dev/null +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/LostConnectionEvent.java @@ -0,0 +1,19 @@ +package org.dna.mqtt.moquette.messaging.spi.impl.events; + +/** + * Used to model the connection lost event + * + * @author andrea + */ +public class LostConnectionEvent extends MessagingEvent{ + private String m_clientID; + + public LostConnectionEvent(String clienID) { + m_clientID = clienID; + } + + public String getClientID() { + return m_clientID; + } + +} diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyMQTTHandler.java b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyMQTTHandler.java index 7e39e1ca..88b8dff5 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyMQTTHandler.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyMQTTHandler.java @@ -10,6 +10,7 @@ import org.dna.mqtt.moquette.proto.Utils; import org.dna.mqtt.moquette.proto.messages.AbstractMessage; import static org.dna.mqtt.moquette.proto.messages.AbstractMessage.*; import org.dna.mqtt.moquette.proto.messages.PingRespMessage; +import org.dna.mqtt.moquette.server.Constants; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -60,6 +61,9 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter { @Override public void channelInactive(ChannelHandlerContext ctx)throws Exception { + NettyChannel channel = m_channelMapper.get(ctx); + String clientID = (String) channel.getAttribute(Constants.ATTR_CLIENTID); + m_messaging.lostConnection(clientID); ctx.close(/*false*/); synchronized(m_channelMapper) { m_channelMapper.remove(ctx);