Introduced handling of sessionPresent flag in ConnAck msg, now on every conenction is created an empty subscription in ISessionStore

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

@@ -36,4 +36,9 @@ public interface ISessionsStore {
     void wipeSubscriptions(String sessionID);
 
     List<Subscription> listAllSubscriptions();
+
+    /**
+     * @return true iff there are subscriptions persisted with clientID
+     */
+    boolean contains(String clientID);
 }

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

@@ -273,6 +273,11 @@ public class HawtDBPersistentStore implements IMessagesStore, ISessionsStore {
         LOG.debug("retrieveAllSubscriptions returning subs {}", allSubscriptions);
         return allSubscriptions;
     }
+    
+    @Override
+    public boolean contains(String clientID) {
+        return m_persistentSubscriptions.containsKey(clientID);
+    }
 
     public void close() {
         LOG.debug("closing disk storage");

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

@@ -134,6 +134,11 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore {
     public void wipeSubscriptions(String clientID) {
         m_persistentSubscriptions.remove(clientID);
     }
+    
+    @Override
+    public boolean contains(String clientID) {
+        return m_persistentSubscriptions.containsKey(clientID);
+    }
 
     public List<Subscription> listAllSubscriptions() {
         List<Subscription> allSubscriptions = new ArrayList<Subscription>();

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

@@ -220,10 +220,16 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
 
         ConnAckMessage okResp = new ConnAckMessage();
         okResp.setReturnCode(ConnAckMessage.CONNECTION_ACCEPTED);
+        if (!msg.isCleanSession() && m_sessionsStore.contains(msg.getClientID())) {
+            okResp.setSessionPresent(true);
+        }
         LOG.debug("processConnect sent OK ConnAck");
         session.write(okResp);
         LOG.info("Connected client ID <{}> with clean session {}", msg.getClientID(), msg.isCleanSession());
         
+        LOG.info("Create persistent session for clientID {}", msg.getClientID());
+        m_sessionsStore.addNewSubscription(Subscription.createEmptySubscription(msg.getClientID(), true), msg.getClientID()); //null means EmptySubscription
+        
         if (!msg.isCleanSession()) {
             //force the republish of stored QoS1 and QoS2
             republishStored(msg.getClientID());

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

@@ -39,6 +39,13 @@ public class Subscription implements Serializable {
         this.topicFilter = topicFilter;
         this.cleanSession = cleanSession;
     }
+    
+    /**
+     * Factory method for empty subscriptions
+     */
+    public static final Subscription createEmptySubscription(String clientId, boolean cleanSession) {
+        return new Subscription(clientId, "", QOSType.MOST_ONE, cleanSession);
+    }
 
     public String getClientId() {
         return clientId;

broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java

@@ -23,23 +23,25 @@ import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
-import org.dna.mqtt.moquette.messaging.spi.ISessionsStore;
 import org.dna.mqtt.moquette.messaging.spi.IMessagesStore;
+import org.dna.mqtt.moquette.messaging.spi.ISessionsStore;
 import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent;
 import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription;
 import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.SubscriptionsStore;
+import static org.dna.mqtt.moquette.parser.netty.Utils.VERSION_3_1_1;
 import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
 import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
 import org.dna.mqtt.moquette.proto.messages.ConnAckMessage;
 import org.dna.mqtt.moquette.proto.messages.ConnectMessage;
+import org.dna.mqtt.moquette.proto.messages.DisconnectMessage;
 import org.dna.mqtt.moquette.proto.messages.PublishMessage;
 import org.dna.mqtt.moquette.proto.messages.SubAckMessage;
 import org.dna.mqtt.moquette.proto.messages.SubscribeMessage;
 import org.dna.mqtt.moquette.server.ServerChannel;
 import static org.junit.Assert.*;
-import static org.mockito.Mockito.*;
 import org.junit.Before;
 import org.junit.Test;
+import static org.mockito.Mockito.*;
 
 /**
  *
@@ -97,7 +99,9 @@ public class ProtocolProcessorTest {
         }
     } 
     
-    
+    /**
+     * This a synchronous channel that avoid output ring buffer from Processor
+     */
     class MockReceiverChannel implements ServerChannel {
 //        byte m_returnCode;
         AbstractMessage m_receivedMessage;
@@ -247,6 +251,31 @@ public class ProtocolProcessorTest {
     }
     
     
+    @Test
+    public void testConnAckContainsSessionPresentFlag() throws InterruptedException {
+        connMsg = new ConnectMessage();
+        connMsg.setProcotolVersion(VERSION_3_1_1);
+        connMsg.setClientID("CliID");
+        connMsg.setCleanSession(false);
+
+        //Connect a first time
+        m_processor.processConnect(m_session, connMsg);
+        //disconnect
+        m_processor.processDisconnect(m_session, "CliID", false);
+              
+        //Exercise, reconnect
+        MockReceiverChannel firstReceiverSession = new MockReceiverChannel();
+        m_processor.processConnect(firstReceiverSession, connMsg);
+
+        //Verify
+        AbstractMessage recvMsg = firstReceiverSession.getMessage();
+        
+        assertTrue(recvMsg instanceof ConnAckMessage);
+        ConnAckMessage connAckMsg = (ConnAckMessage) recvMsg;
+        assertTrue(connAckMsg.isSessionPresent());
+        assertEquals(ConnAckMessage.CONNECTION_ACCEPTED, connAckMsg.getReturnCode());
+    }
+    
     @Test
     public void testPublish() throws InterruptedException {
         final Subscription subscription = new Subscription(FAKE_CLIENT_ID, 

netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/ConnAckEncoder.java

@@ -30,7 +30,7 @@ class ConnAckEncoder extends DemuxEncoder<ConnAckMessage>{
     protected void encode(ChannelHandlerContext chc, ConnAckMessage message, ByteBuf out) {
         out.writeByte(AbstractMessage.CONNACK << 4);
         out.writeBytes(Utils.encodeRemainingLength(2));
-        out.writeByte(0);
+        out.writeByte(message.isSessionPresent() ? 0x01 : 0x00);
         out.writeByte(message.getReturnCode());
     }
     

parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/ConnAckMessage.java

@@ -29,6 +29,7 @@ public class ConnAckMessage extends AbstractMessage {
     public static final byte NOT_AUTHORIZED = 0x05;
     
     private byte m_returnCode;
+    private boolean sessionPresent;
     
     public ConnAckMessage() {
         m_messageType = AbstractMessage.CONNACK;
@@ -41,5 +42,12 @@ public class ConnAckMessage extends AbstractMessage {
     public void setReturnCode(byte returnCode) {
         this.m_returnCode = returnCode;
     }
-    
+
+    public boolean isSessionPresent() {
+        return this.sessionPresent;
+    }
+
+    public void setSessionPresent(boolean present) {
+        this.sessionPresent = present;
+    }
 }