Merge branch 'master' into remove_content_copy

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)

broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java

@@ -14,5 +14,7 @@ public interface IMessaging {
 
     void disconnect(ServerChannel session);
     
+    void lostConnection(String clientID);
+
     void handleProtocolMessage(ServerChannel session, AbstractMessage msg);
 }

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<PublishEvent> 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<PublishEvent> retrivePersistedPublishes(String clientID) {

broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java

@@ -79,11 +79,11 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
     }
     
     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<ValueEvent> {
 
         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<ValueEvent> {
         LOG.debug("republishStored invoked");
         List<PublishEvent> 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<ValueEvent> {
     }
     
     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
@@ -194,6 +196,9 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
         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) {
             //store the temporary message
@@ -228,7 +233,8 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
     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<ValueEvent> {
         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<ValueEvent> {
         //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<ValueEvent> {
     
     
     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<ValueEvent> {
         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

broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java

@@ -83,6 +83,10 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
         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<ValueEvent> {
 
         } else if (evt instanceof InitEvent) {
             processInit();
+        } else if (evt instanceof LostConnectionEvent) {
+            LostConnectionEvent lostEvt = (LostConnectionEvent) evt;
+            m_processor.proccessConnectionLost(lostEvt.getClientID());
         }
     }
 

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;
+    }
+    
+}

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);
     }
 

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);
         }
 
     }

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

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);

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
 

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 {
@@ -276,8 +279,28 @@ 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();

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;

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()
+

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()