diff --git a/ChangeLog.txt b/ChangeLog.txt index b756c485..d9822c3f 100644 --- a/ChangeLog.txt +++ b/ChangeLog.txt @@ -8,3 +8,4 @@ Version 0.3: Version 0.4: Dropped support for MINA + Fix Issue 15 (bad re-activation of client's subscriptions after a recconection) 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/HawtDBStorageService.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java index cc3c126c..18c2c107 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java @@ -159,7 +159,6 @@ public class HawtDBStorageService implements IStorageService { } public void storePublishForFuture(PublishEvent evt) { - LOG.debug("storePublishForFuture store evt " + evt); List storedEvents; String clientID = evt.getClientID(); if (!m_persistentMessageStore.containsKey(clientID)) { @@ -169,6 +168,8 @@ public class HawtDBStorageService implements IStorageService { } storedEvents.add(evt); m_persistentMessageStore.put(clientID, storedEvents); + LOG.debug(String.format("Stored published message for client <%s> on topic <%s>, content %s", + clientID, evt.getTopic(), new String(evt.getMessage()))); } public List retrivePersistedPublishes(String clientID) { 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 95d39ae5..1c1d988e 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 @@ -79,11 +79,11 @@ class ProtocolProcessor implements EventHandler { } void processConnect(ServerChannel session, ConnectMessage msg) { - LOG.info("processConnect for client " + msg.getClientID()); + LOG.debug("processConnect for client " + msg.getClientID()); if (msg.getProcotolVersion() != 0x03) { ConnAckMessage badProto = new ConnAckMessage(); badProto.setReturnCode(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION); - LOG.info("processConnect sent bad proto ConnAck"); + LOG.warn("processConnect sent bad proto ConnAck"); session.write(badProto); session.close(false); return; @@ -156,8 +156,9 @@ class ProtocolProcessor implements EventHandler { ConnAckMessage okResp = new ConnAckMessage(); okResp.setReturnCode(ConnAckMessage.CONNECTION_ACCEPTED); - LOG.info("processConnect sent OK ConnAck"); + LOG.debug("processConnect sent OK ConnAck"); session.write(okResp); + LOG.info(String.format("Connected client ID <%s> with clean session %s", msg.getClientID(), msg.isCleanSession())); if (!msg.isCleanSession()) { //force the republish of stored QoS1 and QoS2 @@ -169,10 +170,11 @@ class ProtocolProcessor implements EventHandler { LOG.debug("republishStored invoked"); List publishedEvents = m_storageService.retrivePersistedPublishes(clientID); if (publishedEvents == null) { - LOG.debug("republishStored, no stored publish events"); + LOG.info(String.format("No stored messages for client <%s>", clientID)); return; } + LOG.info(String.format("republishing stored messages to client <%s>", clientID)); for (PublishEvent pubEvt : publishedEvents) { sendPublish(pubEvt.getClientID(), pubEvt.getTopic(), pubEvt.getQos(), pubEvt.getMessage(), false, pubEvt.getMessageID()); @@ -180,7 +182,7 @@ class ProtocolProcessor implements EventHandler { } private void processRemoveAllSubscriptions(String clientID) { - LOG.debug("processRemoveAllSubscriptions invoked"); + LOG.info(String.format("cleaning old saved subscriptions for client <%s>", clientID)); subscriptions.removeForClient(clientID); //remove also the messages stored of type QoS1/2 @@ -193,6 +195,9 @@ class ProtocolProcessor implements EventHandler { final AbstractMessage.QOSType qos = evt.getQos(); final ByteBuffer message = evt.getMessage(); boolean retain = evt.isRetain(); + + LOG.info(String.format("Publish recieved from clientID <%s> on topic <%s> with QoS %s", + evt.getClientID(), evt.getTopic(), evt.getQos())); String publishKey = null; if (qos == AbstractMessage.QOSType.LEAST_ONE) { @@ -228,7 +233,8 @@ class ProtocolProcessor implements EventHandler { private void publish2Subscribers(String topic, AbstractMessage.QOSType qos, ByteBuffer 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>", + sub.getClientId(), sub.getTopic(), qos)); if (qos == AbstractMessage.QOSType.MOST_ONE) { //QoS 0 sendPublish(sub.getClientId(), topic, qos, message, false); @@ -264,6 +270,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); } @@ -376,6 +384,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> 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)); + } } /** @@ -398,7 +417,7 @@ class ProtocolProcessor implements EventHandler { void processSubscribe(ServerChannel session, SubscribeMessage msg, String clientID, boolean cleanSession) { - LOG.info(String.format("processSubscribe invoked from client %s with msgID %d", clientID, msg.getMessageID())); + LOG.debug(String.format("processSubscribe invoked from client %s with msgID %d", clientID, msg.getMessageID())); for (SubscribeMessage.Couple req : msg.subscriptions()) { AbstractMessage.QOSType qos = AbstractMessage.QOSType.values()[req.getQos()]; @@ -414,11 +433,14 @@ class ProtocolProcessor implements EventHandler { for (int i = 0; i < msg.subscriptions().size(); i++) { ackMessage.addType(AbstractMessage.QOSType.MOST_ONE); } - LOG.info("replying with SubAck to MSG ID " + msg.getMessageID()); + LOG.debug("replying with SubAck to MSG ID " + msg.getMessageID()); session.write(ackMessage); } private void subscribeSingleTopic(Subscription newSubscription, final String topic) { + LOG.info(String.format("<%s> subscribed to topic <%s> with QoS %s", + newSubscription.getClientId(), topic, + AbstractMessage.QOSType.formatQoS(newSubscription.getRequestedQos()))); subscriptions.add(newSubscription); //scans retained messages to be published to the new subscription 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 354671fe..c9238085 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)); @@ -151,6 +155,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/messaging/spi/impl/subscriptions/SubscriptionsStore.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStore.java index e57beec6..3871d843 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStore.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStore.java @@ -166,7 +166,7 @@ public class SubscriptionsStore { } public void activate(String clientID) { - LOG.debug("activate re-activating subscriptions for clientID " + clientID); + LOG.debug(String.format("Activating subscriptions for clientID <%s>", clientID)); subscriptions.activate(clientID); } diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/TreeNode.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/TreeNode.java index e905d9ee..1dcbc2ef 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/TreeNode.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/TreeNode.java @@ -148,7 +148,7 @@ class TreeNode { //go deep for (TreeNode child : m_children) { - child.deactivate(clientID); + child.activate(clientID); } } diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/Server.java b/broker/src/main/java/org/dna/mqtt/moquette/server/Server.java index 4fcd9953..78e230dd 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/server/Server.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/server/Server.java @@ -25,6 +25,7 @@ public class Server { final Server server = new Server(); server.startServer(); + System.out.println("Server started, version 0.4-SNAPSHOT"); //Bind a shutdown hook Runtime.getRuntime().addShutdownHook(new Thread() { @Override 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); diff --git a/broker/src/main/resources/log4j.properties b/broker/src/main/resources/log4j.properties index 1b51ddd9..db932b00 100644 --- a/broker/src/main/resources/log4j.properties +++ b/broker/src/main/resources/log4j.properties @@ -12,10 +12,15 @@ log4j.rootLogger=ERROR, stdout, file log4j.logger.org.dna.mqtt=INFO log4j.logger.org.dna.mqtt.moquette.server.ServerIntegrationFuseTest=DEBUG log4j.logger.org.dna.mqtt.moquette.messaging.spi.impl.ProtocolProcessor=INFO +log4j.logger.org.dna.mqtt.moquette.messaging.spi.impl.SimpleMessaging=WARN +#Protocol parsing +log4j.logger.org.dna.mqtt.moquette.server.netty.NettyMQTTHandler=WARN +log4j.logger.org.dna.mqtt.moquette.parser.netty=WARN # stdout appender is set to be a ConsoleAppender. log4j.appender.stdout=org.apache.log4j.ConsoleAppender -log4j.appender.stdout.Threshold=WARN +#log4j.appender.stdout.Threshold=WARN +log4j.appender.stdout.Threshold=INFO log4j.appender.stdout.layout=org.apache.log4j.PatternLayout log4j.appender.stdout.layout.ConversionPattern=%-4r [%t] %-5p %c{1} %x - %m%n diff --git a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationPahoTest.java b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationPahoTest.java index 7703813b..151fb330 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationPahoTest.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationPahoTest.java @@ -62,6 +62,8 @@ public class ServerIntegrationPahoTest { private static final Logger LOG = LoggerFactory.getLogger(ServerIntegrationPahoTest.class); static MqttClientPersistence s_dataStore; + static MqttClientPersistence s_pubDataStore; + Server m_server; IMqttClient m_client; @@ -71,6 +73,7 @@ public class ServerIntegrationPahoTest { public static void beforeTests() { String tmpDir = System.getProperty("java.io.tmpdir"); s_dataStore = new MqttDefaultFilePersistence(tmpDir); + s_pubDataStore = new MqttDefaultFilePersistence(tmpDir + File.separator + "publisher"); } protected void startServer() throws IOException { @@ -275,9 +278,29 @@ public class ServerIntegrationPahoTest { assertEquals("Hello MQTT", m_callback.getMessage().toString()); } - + + @Test + public void checkReceivePublishedMessage_after_a_reconnect_with_notCleanSession() throws Exception { + LOG.info("*** checkReceivePublishedMessage_after_a_reconnect_with_notCleanSession ***"); + MqttConnectOptions options = new MqttConnectOptions(); + options.setCleanSession(false); + m_client.connect(options); + m_client.subscribe("/topic", 1); + m_client.disconnect(); + + m_client.connect(options); + m_client.subscribe("/topic", 1); + + //publish a QoS 1 message another client publish a message on the topic + publishFromAnotherClient("/topic", "Hello MQTT".getBytes(), 1); + + //Verify that after a reconnection the client receive the message + assertNotNull(m_callback.getMessage()); + assertEquals("Hello MQTT", m_callback.getMessage().toString()); + } + private void publishFromAnotherClient(String topic, byte[] payload, int qos) throws Exception { - IMqttClient anotherClient = new MqttClient("tcp://localhost:1883", "TestClientPUB", s_dataStore); + IMqttClient anotherClient = new MqttClient("tcp://localhost:1883", "TestClientPUB", s_pubDataStore); anotherClient.connect(); anotherClient.publish(topic, payload, qos, false); anotherClient.disconnect(); diff --git a/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/AbstractMessage.java b/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/AbstractMessage.java index 61e1c332..0f9b0ac7 100644 --- a/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/AbstractMessage.java +++ b/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/AbstractMessage.java @@ -24,6 +24,10 @@ public abstract class AbstractMessage { public static enum QOSType { MOST_ONE, LEAST_ONE, EXACTLY_ONCE, RESERVED; + + public static String formatQoS(QOSType qos) { + return String.format("%d - %s", qos.ordinal(), qos.name()); + } } //type protected boolean m_dupFlag; diff --git a/tools_scripts/issue15_receiver.groovy b/tools_scripts/issue15_receiver.groovy new file mode 100644 index 00000000..a2900e2b --- /dev/null +++ b/tools_scripts/issue15_receiver.groovy @@ -0,0 +1,48 @@ +@Grab(group='org.fusesource.mqtt-client', module='mqtt-client', version='1.5') + +import java.net.URISyntaxException + +import org.fusesource.mqtt.client.BlockingConnection +import org.fusesource.mqtt.client.MQTT +import org.fusesource.mqtt.client.Message +import org.fusesource.mqtt.client.QoS +import org.fusesource.mqtt.client.Topic + +class Reciever { + MQTT mqtt = new MQTT() + BlockingConnection connection + + public void connectAndSubscribe(boolean cleanSession) { + mqtt.setCleanSession(cleanSession) + mqtt.setHost("localhost", 1883) + + mqtt.setClientId("reciever") + connection = mqtt.blockingConnection() + connection.connect() + Topic[] topics = [new Topic("/topic", QoS.AT_LEAST_ONCE)] + byte[] qoses = connection.subscribe(topics) + } +} + + +Reciever r = new Reciever() +r.connectAndSubscribe(false) +Message message +3.times { i -> + message = r.connection.receive() + println "Topic: ${message.topic}, payload: ${new String(message.payload)}" + message.ack() +} +r.connection.kill() +//r.connection.disconnect() + +Thread.sleep(3000) + +r.connectAndSubscribe(false) +3.times { i -> + message = r.connection.receive() + println "Topic: ${message.topic}, payload: ${new String(message.payload)}" + message.ack() +} +r.connection.disconnect() + diff --git a/tools_scripts/issue15_sender.groovy b/tools_scripts/issue15_sender.groovy new file mode 100644 index 00000000..f79d4a36 --- /dev/null +++ b/tools_scripts/issue15_sender.groovy @@ -0,0 +1,27 @@ +//@GrabResolver(name='Paho', root='https://repo.eclipse.org/content/repositories/paho-releases/') +@Grab(group='org.fusesource.mqtt-client', module='mqtt-client', version='1.5') + +import java.net.URISyntaxException + +import org.fusesource.mqtt.client.BlockingConnection +import org.fusesource.mqtt.client.MQTT +import org.fusesource.mqtt.client.QoS + + +MQTT mqtt = new MQTT() +//mqtt.setHost("test.mosquitto.org", 1883); +mqtt.setHost("localhost", 1883) + +BlockingConnection connection = mqtt.blockingConnection() +connection.connect() + +//for (int i = 0; i < 1000; i++) { +(1..1000).each{ i -> +// LOG.info("Publishing"); + String payload = "Hello world MQTT!!${i}" + connection.publish("/topic", payload.getBytes(), QoS.AT_LEAST_ONCE, false) + println payload + Thread.sleep 1000 +} + +connection.disconnect()