From 482ffe0c651d8e5fe3aa7e60de09aa30b7750125 Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Sun, 27 Jul 2014 12:20:53 +0200 Subject: [PATCH] Introduced handling of sessionPresent flag in ConnAck msg, now on every conenction is created an empty subscription in ISessionStore --- .../messaging/spi/ISessionsStore.java | 5 +++ .../spi/impl/HawtDBPersistentStore.java | 5 +++ .../spi/impl/MemoryStorageService.java | 5 +++ .../messaging/spi/impl/ProtocolProcessor.java | 6 ++++ .../spi/impl/subscriptions/Subscription.java | 7 ++++ .../spi/impl/ProtocolProcessorTest.java | 35 +++++++++++++++++-- .../moquette/parser/netty/ConnAckEncoder.java | 2 +- .../proto/messages/ConnAckMessage.java | 10 +++++- 8 files changed, 70 insertions(+), 5 deletions(-) diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/ISessionsStore.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/ISessionsStore.java index ad4add5b..6926aa0a 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/ISessionsStore.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/ISessionsStore.java @@ -36,4 +36,9 @@ public interface ISessionsStore { void wipeSubscriptions(String sessionID); List listAllSubscriptions(); + + /** + * @return true iff there are subscriptions persisted with clientID + */ + boolean contains(String clientID); } diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBPersistentStore.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBPersistentStore.java index 905c9666..fd6478a5 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBPersistentStore.java +++ b/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"); diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java index cf6596b3..61dd8531 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java +++ b/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 listAllSubscriptions() { List allSubscriptions = new ArrayList(); 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 2e9393f5..537be8c4 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 @@ -220,10 +220,16 @@ class ProtocolProcessor implements EventHandler { 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()); diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/Subscription.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/Subscription.java index cbc96858..4d354d4e 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/Subscription.java +++ b/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; diff --git a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java index 02da3116..050aa9d4 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java +++ b/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, diff --git a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/ConnAckEncoder.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/ConnAckEncoder.java index 856aa4ba..41ef0c56 100644 --- a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/ConnAckEncoder.java +++ b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/ConnAckEncoder.java @@ -30,7 +30,7 @@ class ConnAckEncoder extends DemuxEncoder{ 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()); } diff --git a/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/ConnAckMessage.java b/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/ConnAckMessage.java index 00f5dae2..7b898452 100644 --- a/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/ConnAckMessage.java +++ b/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; + } }