From 6e1524817eba484d39eced6913dd96a7def0cafd Mon Sep 17 00:00:00 2001 From: andsel Date: Sun, 20 Sep 2015 14:10:28 +0200 Subject: [PATCH] Fixed ugly empty session storage, avoid fake NullObject empty subscription subscription --- ChangeLog.txt | 1 + .../eclipse/moquette/spi/ISessionsStore.java | 2 ++ .../spi/impl/MemoryStorageService.java | 9 +++++ .../moquette/spi/impl/ProtocolProcessor.java | 9 +++-- .../spi/persistence/MapDBPersistentStore.java | 13 ++++++- .../impl/ProtocolProcessor_CONNECT_Test.java | 36 ++++++++++++++++--- 6 files changed, 62 insertions(+), 8 deletions(-) diff --git a/ChangeLog.txt b/ChangeLog.txt index 7841735f..9350db12 100644 --- a/ChangeLog.txt +++ b/ChangeLog.txt @@ -8,6 +8,7 @@ Version 0.8: Moved config to be not only filesystem based, but also classpath and memory (fix issue #73) Switched from commit pesistence on every change to time based to fix delay in handling many msgs/sec (fix issue #79) Changed willMessage and password in CONNECT message from String to byte[] (fix issue #94 and #95) + Fixed ugly empty session storage, avoid fake NullObject empty subscription subscription Version 0.7: Fix issue #38, added aligned to MQTT 3.1.1 diff --git a/broker/src/main/java/org/eclipse/moquette/spi/ISessionsStore.java b/broker/src/main/java/org/eclipse/moquette/spi/ISessionsStore.java index db824c2a..1c5e8b29 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/ISessionsStore.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/ISessionsStore.java @@ -53,4 +53,6 @@ public interface ISessionsStore { * @return true iff there are subscriptions persisted with clientID */ boolean contains(String clientID); + + void createNewSession(String clientID); } diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java index d69bc54d..0448dde7 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java @@ -208,6 +208,15 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore { return m_persistentSubscriptions.containsKey(clientID); } + @Override + public void createNewSession(String clientID) { + if (m_persistentSubscriptions.containsKey(clientID)) { + LOG.error("already exists a session for client <{}>", clientID); + return; + } + m_persistentSubscriptions.put(clientID, new HashSet()); + } + @Override public List listAllSubscriptions() { List allSubscriptions = new ArrayList(); diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java index 2b50e11c..9f40542c 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java @@ -243,14 +243,17 @@ class ProtocolProcessor implements EventHandler { ConnAckMessage okResp = new ConnAckMessage(); okResp.setReturnCode(ConnAckMessage.CONNECTION_ACCEPTED); - if (!msg.isCleanSession() && m_sessionsStore.contains(msg.getClientID())) { + boolean isSessionAlreadyStored = m_sessionsStore.contains(msg.getClientID()); + if (!msg.isCleanSession() && isSessionAlreadyStored) { okResp.setSessionPresent(true); } session.write(okResp); m_interceptor.notifyClientConnected(msg); - LOG.info("Create persistent session for clientID <{}>", msg.getClientID()); - m_sessionsStore.addNewSubscription(Subscription.createEmptySubscription(msg.getClientID(), true)); //null means EmptySubscription + if (!isSessionAlreadyStored) { + LOG.info("Create persistent session for clientID <{}>", msg.getClientID()); + m_sessionsStore.createNewSession(msg.getClientID()); + } LOG.info("Connected client ID <{}> with clean session {}", msg.getClientID(), msg.isCleanSession()); if (!msg.isCleanSession()) { //force the republish of stored QoS1 and QoS2 diff --git a/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java b/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java index 61010414..2de09f82 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java @@ -285,7 +285,7 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { } public List listAllSubscriptions() { - List allSubscriptions = new ArrayList(); + List allSubscriptions = new ArrayList<>(); for (Map.Entry> entry : m_persistentSubscriptions.entrySet()) { allSubscriptions.addAll(entry.getValue()); } @@ -298,6 +298,17 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { return m_persistentSubscriptions.containsKey(clientID); } + @Override + public void createNewSession(String clientID) { + LOG.debug("createNewSession for client <{}>", clientID); + if (m_persistentSubscriptions.containsKey(clientID)) { + LOG.error("already exists a session for client <{}>", clientID); + return; + } + LOG.debug("clientID {} is a newcome, creating it's empty subscriptions set", clientID); + m_persistentSubscriptions.put(clientID, new HashSet()); + } + public void close() { this.m_db.commit(); LOG.debug("persisted subscriptions {}", m_persistentSubscriptions); diff --git a/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java b/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java index aa791269..2a51a99e 100644 --- a/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java +++ b/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java @@ -15,14 +15,12 @@ */ package org.eclipse.moquette.spi.impl; -import org.eclipse.moquette.proto.messages.AbstractMessage; -import org.eclipse.moquette.proto.messages.ConnAckMessage; -import org.eclipse.moquette.proto.messages.ConnectMessage; -import org.eclipse.moquette.proto.messages.DisconnectMessage; +import org.eclipse.moquette.proto.messages.*; import org.eclipse.moquette.server.netty.NettyChannel; import org.eclipse.moquette.spi.IMessagesStore; import org.eclipse.moquette.spi.ISessionsStore; import org.eclipse.moquette.spi.impl.security.PermitAllAuthorizator; +import org.eclipse.moquette.spi.impl.subscriptions.Subscription; import org.eclipse.moquette.spi.impl.subscriptions.SubscriptionsStore; import org.junit.Before; import org.junit.Test; @@ -258,4 +256,34 @@ public class ProtocolProcessor_CONNECT_Test { assertEquals(ConnAckMessage.CONNECTION_ACCEPTED, connAckMsg.getReturnCode()); } + + @Test + public void testMultipleReconnection() throws InterruptedException { + //connect with clean a false and subscribe to a topic + connMsg = new ConnectMessage(); + connMsg.setProtocolVersion(VERSION_3_1_1); + connMsg.setClientID("CliID"); + connMsg.setCleanSession(false); + m_processor.processConnect(m_session, connMsg); + assertEquals(ConnAckMessage.CONNECTION_ACCEPTED, m_session.getReturnCode()); + + //subscribe + SubscribeMessage subscribeMsg = new SubscribeMessage(); + subscribeMsg.addSubscription(new SubscribeMessage.Couple((byte) AbstractMessage.QOSType.MOST_ONE.ordinal(), FAKE_TOPIC)); + m_session.setAttribute(NettyChannel.ATTR_KEY_CLIENTID, "CliID"); + m_session.setAttribute(NettyChannel.ATTR_KEY_CLEANSESSION, false); + m_processor.processSubscribe(m_session, subscribeMsg); + Subscription expectedSubscription = new Subscription("CliID", FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, false); + assertTrue(subscriptions.contains(expectedSubscription)); + + //disconnect + m_processor.processDisconnect(m_session, new DisconnectMessage()); + + //reconnect clean session a false + m_processor.processConnect(m_session, connMsg); + assertEquals(ConnAckMessage.CONNECTION_ACCEPTED, m_session.getReturnCode()); + + //verify that the first subscription is still preserved + assertTrue(subscriptions.contains(expectedSubscription)); + } }