diff --git a/ChangeLog.txt b/ChangeLog.txt index a9c05696..3a0b2615 100644 --- a/ChangeLog.txt +++ b/ChangeLog.txt @@ -1,7 +1,8 @@ Version 0.9: Moved canRead check to topic from publish phase to subscribe (#144) Added javadoc and src jars creations for JCenter/Maven central - + Added persisting of clean session flag for already existing client session (fix issue #145) + Version 0.8: Introduced authenticator and authorizator implementations. Fixed bug, doesn't close the channel when connect doesn't contains a valid proto name, issue #65 diff --git a/broker/src/main/java/io/moquette/spi/ClientSession.java b/broker/src/main/java/io/moquette/spi/ClientSession.java index f76d04f3..2be43456 100644 --- a/broker/src/main/java/io/moquette/spi/ClientSession.java +++ b/broker/src/main/java/io/moquette/spi/ClientSession.java @@ -55,7 +55,7 @@ public class ClientSession { private Set subscriptions = new HashSet<>(); - public final boolean cleanSession; + private volatile boolean cleanSession; private boolean active = false; @@ -146,6 +146,11 @@ public class ClientSession { return this.cleanSession; } + public void cleanSession(boolean cleanSession) { + this.cleanSession = cleanSession; + this.m_sessionsStore.updateCleanStatus(this.clientID, cleanSession); + } + public void activate() { this.active = true; } diff --git a/broker/src/main/java/io/moquette/spi/ISessionsStore.java b/broker/src/main/java/io/moquette/spi/ISessionsStore.java index ed894266..bee5d133 100644 --- a/broker/src/main/java/io/moquette/spi/ISessionsStore.java +++ b/broker/src/main/java/io/moquette/spi/ISessionsStore.java @@ -27,6 +27,8 @@ import io.moquette.spi.impl.subscriptions.Subscription; */ public interface ISessionsStore { + void updateCleanStatus(String clientID, boolean cleanSession); + class ClientTopicCouple { public final String topicFilter; public final String clientID; @@ -99,6 +101,7 @@ public interface ISessionsStore { ClientSession createNewSession(String clientID, boolean cleanSession); /** + * @param clientID the client owning the session. * @return the session for the given clientID, null if not found. * */ ClientSession sessionForClient(String clientID); diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java index 54575880..a16d2826 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -219,6 +219,9 @@ public class ProtocolProcessor { if (!msg.isCleanSession() && isSessionAlreadyStored) { okResp.setSessionPresent(true); } + if (isSessionAlreadyStored) { + clientSession.cleanSession(msg.isCleanSession()); + } channel.writeAndFlush(okResp); m_interceptor.notifyClientConnected(msg); diff --git a/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java b/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java index 50848f0c..5e82ba95 100644 --- a/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java +++ b/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java @@ -143,6 +143,11 @@ class MapDBSessionsStore implements ISessionsStore { return new ClientSession(clientID, m_messagesStore, this, storedSession.cleanSession); } + @Override + public void updateCleanStatus(String clientID, boolean cleanSession) { + m_persistentSessions.put(clientID, new MapDBPersistentStore.PersistentSession(cleanSession)); + } + /** * Return the next valid packetIdentifier for the given client session. * */ diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java index 830089a4..f8d94203 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java @@ -19,6 +19,7 @@ import io.moquette.proto.messages.AbstractMessage; import io.moquette.proto.messages.ConnectMessage; import io.moquette.proto.messages.SubscribeMessage; import io.moquette.server.netty.NettyUtils; +import io.moquette.spi.ClientSession; import io.moquette.spi.IMessagesStore; import io.moquette.spi.ISessionsStore; import io.moquette.spi.impl.subscriptions.SubscriptionsStore; @@ -207,6 +208,35 @@ public class ProtocolProcessor_CONNECT_Test { assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound()); } + @Test + public void connectWithCleanSessionUpdateClientSession() throws InterruptedException { + m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, true, + new PermitAllAuthorizator(), ProtocolProcessorTest.NO_OBSERVERS_INTERCEPTOR); + + //first connect with clean session true + connMsg.setClientID("123"); + connMsg.setCleanSession(true); + m_processor.processConnect(m_session, connMsg); + assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound()); + m_processor.processDisconnect(m_session); + assertFalse(m_session.isOpen()); + + //second connect with clean session false + m_session = new EmbeddedChannel(); + ConnectMessage secondConnMsg = new ConnectMessage(); + secondConnMsg.setProtocolVersion((byte) 0x03); + secondConnMsg.setClientID("123"); + m_processor.processConnect(m_session, secondConnMsg); + assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound()); + + //Verify client session is clean false + ClientSession session = m_sessionStore.sessionForClient("123"); + assertFalse(session.isCleanSession()); + + //Verify + //assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound()); + } + @Test public void connectWithSameClientIDBadCredentialsDoesntDropExistingClient() { //Connect a client1 diff --git a/broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java b/broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java index 731cbb4b..1a228198 100644 --- a/broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java +++ b/broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java @@ -125,6 +125,11 @@ public class MemorySessionStore implements ISessionsStore { return new ClientSession(clientID, m_messagesStore, this, storedSession.cleanSession); } + @Override + public void updateCleanStatus(String clientID, boolean cleanSession) { + m_persistentSessions.put(clientID, new MapDBPersistentStore.PersistentSession(cleanSession)); + } + @Override public List listAllSubscriptions() { List allSubscriptions = new ArrayList<>();