diff --git a/ChangeLog.txt b/ChangeLog.txt index 414ca6e8..579ee1b7 100644 --- a/ChangeLog.txt +++ b/ChangeLog.txt @@ -16,6 +16,7 @@ Version 0.8: Added config property 'autosave_interval' (issue #97) Removed leaking ref to parent's node (issue #119 and #126) Added internal publish once the broker is embedded (issue #113) + Removed client active/not active state from persistence (issue #118) Version 0.7: Fix issue #38, added aligned to MQTT 3.1.1 diff --git a/broker/src/main/java/io/moquette/spi/ClientSession.java b/broker/src/main/java/io/moquette/spi/ClientSession.java index e977ef20..9ebd6f75 100644 --- a/broker/src/main/java/io/moquette/spi/ClientSession.java +++ b/broker/src/main/java/io/moquette/spi/ClientSession.java @@ -148,12 +148,10 @@ public class ClientSession { public void activate() { this.active = true; - this.m_sessionsStore.activate(this.clientID); } public void deactivate() { this.active = false; - this.m_sessionsStore.deactivate(this.clientID); } public boolean isActive() { diff --git a/broker/src/main/java/io/moquette/spi/ISessionsStore.java b/broker/src/main/java/io/moquette/spi/ISessionsStore.java index 9308e0e1..99af0908 100644 --- a/broker/src/main/java/io/moquette/spi/ISessionsStore.java +++ b/broker/src/main/java/io/moquette/spi/ISessionsStore.java @@ -103,10 +103,6 @@ public interface ISessionsStore { * */ ClientSession sessionForClient(String clientID); - void activate(String clientID); - - void deactivate(String clientID); - void inFlightAck(String clientID, int messageID); /** 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 28eeb93f..0fd508b0 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -92,7 +92,7 @@ public class ProtocolProcessor { private static final Logger LOG = LoggerFactory.getLogger(ProtocolProcessor.class); - private Map m_clientIDs; + protected Map m_clientIDs; private SubscriptionsStore subscriptions; private boolean allowAnonymous; private IAuthorizator m_authorizator; @@ -264,9 +264,16 @@ public class ProtocolProcessor { int messageID = msg.getMessageID(); //Remove the message from message store ClientSession targetSession = m_sessionsStore.sessionForClient(clientID); + verifyToActivate(clientID, targetSession); targetSession.inFlightAcknowledged(messageID); } + private void verifyToActivate(String clientID, ClientSession targetSession) { + if (m_clientIDs.containsKey(clientID)) { + targetSession.activate(); + } + } + private static IMessagesStore.StoredMessage asStoredMessage(PublishMessage msg) { IMessagesStore.StoredMessage stored = new IMessagesStore.StoredMessage(msg.getPayload().array(), msg.getQos(), msg.getTopicName()); stored.setRetained(msg.isRetainFlag()); @@ -452,6 +459,7 @@ public class ProtocolProcessor { qos = sub.getRequestedQos(); } ClientSession targetSession = m_sessionsStore.sessionForClient(sub.getClientId()); + verifyToActivate(sub.getClientId(), targetSession); LOG.debug("Broker republishing to client <{}> topic <{}> qos <{}>, active {}", sub.getClientId(), sub.getTopicFilter(), qos, targetSession.isActive()); @@ -554,6 +562,7 @@ public class ProtocolProcessor { int messageID = msg.getMessageID(); LOG.debug("PUB --PUBREL--> SRV processPubRel invoked for clientID {} ad messageID {}", clientID, messageID); ClientSession targetSession = m_sessionsStore.sessionForClient(clientID); + verifyToActivate(clientID, targetSession); IMessagesStore.StoredMessage evt = targetSession.storedMessage(messageID); route2Subscribers(evt); @@ -581,6 +590,7 @@ public class ProtocolProcessor { String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID); int messageID = msg.getMessageID(); ClientSession targetSession = m_sessionsStore.sessionForClient(clientID); + verifyToActivate(clientID, targetSession); //remove from the inflight and move to the QoS2 second phase queue targetSession.inFlightAcknowledged(messageID); targetSession.secondPhaseAckWaiting(messageID); @@ -599,6 +609,7 @@ public class ProtocolProcessor { LOG.debug("\t\tSRV <--PUBCOMP-- SUB processPubComp invoked for clientID {} ad messageID {}", clientID, messageID); //once received the PUBCOMP then remove the message from the temp memory ClientSession targetSession = m_sessionsStore.sessionForClient(clientID); + verifyToActivate(clientID, targetSession); targetSession.secondPhaseAcknowledged(messageID); } @@ -646,7 +657,7 @@ public class ProtocolProcessor { LOG.debug("UNSUBSCRIBE subscription on topics {} for clientID <{}>", topics, clientID); ClientSession clientSession = m_sessionsStore.sessionForClient(clientID); - + verifyToActivate(clientID, clientSession); for (String topic : topics) { boolean validTopic = SubscriptionsStore.validate(topic); if (!validTopic) { @@ -674,6 +685,7 @@ public class ProtocolProcessor { LOG.debug("SUBSCRIBE client <{}> packetID {}", clientID, msg.getMessageID()); ClientSession clientSession = m_sessionsStore.sessionForClient(clientID); + verifyToActivate(clientID, clientSession); //ack the client SubAckMessage ackMessage = new SubAckMessage(); ackMessage.setMessageID(msg.getMessageID()); @@ -715,7 +727,7 @@ public class ProtocolProcessor { }); ClientSession targetSession = m_sessionsStore.sessionForClient(newSubscription.getClientId()); - + verifyToActivate(newSubscription.getClientId(), targetSession); for (IMessagesStore.StoredMessage storedMsg : messages) { //fire the as retained the message LOG.debug("send publish message for topic {}", newSubscription.getTopicFilter()); diff --git a/broker/src/main/java/io/moquette/spi/persistence/MapDBPersistentStore.java b/broker/src/main/java/io/moquette/spi/persistence/MapDBPersistentStore.java index 4d355990..45ecfedc 100644 --- a/broker/src/main/java/io/moquette/spi/persistence/MapDBPersistentStore.java +++ b/broker/src/main/java/io/moquette/spi/persistence/MapDBPersistentStore.java @@ -46,11 +46,9 @@ public class MapDBPersistentStore { * */ public static class PersistentSession implements Serializable { public final boolean cleanSession; - public final boolean active; - public PersistentSession(boolean cleanSession, boolean active) { + public PersistentSession(boolean cleanSession) { this.cleanSession = cleanSession; - this.active = active; } } 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 728a1f27..1d019446 100644 --- a/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java +++ b/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java @@ -129,7 +129,7 @@ class MapDBSessionsStore implements ISessionsStore { throw new IllegalArgumentException("Can't create a session with the ID of an already existing" + clientID); } LOG.debug("clientID {} is a newcome, creating it's empty subscriptions set", clientID); - m_persistentSessions.putIfAbsent(clientID, new PersistentSession(cleanSession, false)); + m_persistentSessions.putIfAbsent(clientID, new PersistentSession(cleanSession)); return new ClientSession(clientID, m_messagesStore, this, cleanSession); } @@ -140,30 +140,7 @@ class MapDBSessionsStore implements ISessionsStore { } PersistentSession storedSession = m_persistentSessions.get(clientID); - ClientSession clientSession = new ClientSession(clientID, m_messagesStore, this, storedSession.cleanSession); - if (storedSession.active) { - clientSession.activate(); - } - return clientSession; - } - - @Override - public void activate(String clientID) { - activationHelper(clientID, true); - } - - @Override - public void deactivate(String clientID) { - activationHelper(clientID, false); - } - - private void activationHelper(String clientID, boolean activation) { - PersistentSession storedSession = m_persistentSessions.get(clientID); - if (storedSession == null) { - throw new IllegalStateException((activation ? "activating" : "deactivating") + " a session never stored/created, clientID <"+ clientID + ">", null); - } - PersistentSession newStoredSession = new PersistentSession(storedSession.cleanSession, activation); - m_persistentSessions.put(clientID, newStoredSession); + return new ClientSession(clientID, m_messagesStore, this, storedSession.cleanSession); } @Override diff --git a/broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java index 5ab9acad..07436b12 100644 --- a/broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java +++ b/broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java @@ -401,4 +401,46 @@ public class ServerIntegrationPahoTest { subscriberB.disconnect(); } + @Test + public void testSubcriptionDoesntStayActiveAfterARestart() throws Exception { + LOG.info("*** testSubcriptionDoesntStayActiveAfterARestart ***"); + String tmpDir = System.getProperty("java.io.tmpdir"); + //clientForSubscribe1 connect and subscribe to /topic QoS2 + MqttClientPersistence dsSubscriberA = new MqttDefaultFilePersistence(tmpDir + File.separator + "clientForSubscribe1"); + + MqttClient clientForSubscribe1 = new MqttClient("tcp://localhost:1883", "clientForSubscribe1", dsSubscriberA); + TestCallback cbSubscriber1 = new TestCallback(); + clientForSubscribe1.setCallback(cbSubscriber1); + clientForSubscribe1.connect(); + clientForSubscribe1.subscribe("topic", 0); + + //server stop + m_server.stopServer(); + System.out.println("\n\n SEVER REBOOTING \n\n"); + //server start + startServer(); + + //clientForSubscribe2 connect and subscribe to /topic QoS2 + MqttClientPersistence dsSubscriberB = new MqttDefaultFilePersistence(tmpDir + File.separator + "clientForSubscribe2"); + MqttClient clientForSubscribe2 = new MqttClient("tcp://localhost:1883", "clientForSubscribe2", dsSubscriberB); + TestCallback cbSubscriber2 = new TestCallback(); + clientForSubscribe2.setCallback(cbSubscriber2); + clientForSubscribe2.connect(); + clientForSubscribe2.subscribe("topic", 0); + + //clientForPublish publish on /topic with QoS2 a message + MqttClientPersistence dsSubscriberPUB = new MqttDefaultFilePersistence(tmpDir + File.separator + "clientForPublish"); + MqttClient clientForPublish = new MqttClient("tcp://localhost:1883", "clientForPublish", dsSubscriberPUB); + clientForPublish.connect(); + clientForPublish.publish("topic", "Hello".getBytes(), 2, true); + + //verify clientForSubscribe1 doesn't receive a notification but clientForSubscribe2 yes + System.out.println("Before waiting to receive 1 sec from " + clientForSubscribe1.getClientId()); + assertFalse(clientForSubscribe1.isConnected()); + assertTrue(clientForSubscribe2.isConnected()); + System.out.println("Waiting to receive 1 sec from " + clientForSubscribe2.getClientId()); + MqttMessage messageOnB = cbSubscriber2.getMessage(true); + assertEquals("Hello", new String(messageOnB.getPayload())); + } + } diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java index f3f79fcc..659233f2 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java @@ -18,6 +18,7 @@ package io.moquette.spi.impl; import io.moquette.proto.messages.*; import io.moquette.interception.InterceptHandler; import io.moquette.proto.messages.AbstractMessage.QOSType; +import io.moquette.server.ConnectionDescriptor; import io.moquette.server.netty.NettyChannel; import io.moquette.spi.ClientSession; import io.moquette.spi.IMatchingCondition; @@ -504,6 +505,9 @@ public class ProtocolProcessorTest { } }; processor.init(subscriptions, memoryMessageStore, sessionsStore, null, true, null, NO_OBSERVERS_INTERCEPTOR); + //just to activate the two sessions + processor.m_clientIDs.put("Sub A", new ConnectionDescriptor("Sub A", null, true)); + processor.m_clientIDs.put("Sub B", new ConnectionDescriptor("Sub B", null, true)); //Exercise processor.route2Subscribers(forwardPublish); 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 fdda3672..0d21f1a6 100644 --- a/broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java +++ b/broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java @@ -110,7 +110,7 @@ public class MemorySessionStore implements ISessionsStore { } LOG.debug("clientID {} is a newcome, creating it's empty subscriptions set", clientID); m_persistentSubscriptions.put(clientID, new HashSet()); - m_persistentSessions.put(clientID, new MapDBPersistentStore.PersistentSession(cleanSession, false)); + m_persistentSessions.put(clientID, new MapDBPersistentStore.PersistentSession(cleanSession)); return new ClientSession(clientID, m_messagesStore, this, cleanSession); } @@ -121,11 +121,7 @@ public class MemorySessionStore implements ISessionsStore { } MapDBPersistentStore.PersistentSession storedSession = m_persistentSessions.get(clientID); - ClientSession clientSession = new ClientSession(clientID, m_messagesStore, this, storedSession.cleanSession); - if (storedSession.active) { - clientSession.activate(); - } - return clientSession; + return new ClientSession(clientID, m_messagesStore, this, storedSession.cleanSession); } @Override @@ -153,25 +149,6 @@ public class MemorySessionStore implements ISessionsStore { return null; } - @Override - public void activate(String clientID) { - activationHelper(clientID, true); - } - - @Override - public void deactivate(String clientID) { - activationHelper(clientID, false); - } - - private void activationHelper(String clientID, boolean activation) { - MapDBPersistentStore.PersistentSession storedSession = m_persistentSessions.get(clientID); - if (storedSession == null) { - throw new IllegalStateException((activation ? "activating" : "deactivating") + " a session never stored/created, clientID <"+ clientID + ">", null); - } - MapDBPersistentStore.PersistentSession newStoredSession = new MapDBPersistentStore.PersistentSession(storedSession.cleanSession, activation); - m_persistentSessions.put(clientID, newStoredSession); - } - @Override public void inFlightAck(String clientID, int messageID) { Map m = this.m_inflightStore.get(clientID);