From dfa75e145fe8a7fc9266826ab5ba0e3288992a6a Mon Sep 17 00:00:00 2001 From: andsel Date: Mon, 25 Jan 2016 22:14:12 +0100 Subject: [PATCH] Fixed bug #143 avoid publication try to disconnected clients --- .../spi/impl/subscriptions/SubscriptionsStore.java | 4 ++++ .../ServerIntegrationEmbeddedPublishTest.java | 14 ++++++++++++++ 2 files changed, 18 insertions(+) diff --git a/broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionsStore.java b/broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionsStore.java index 2fce9744..6330f9a2 100644 --- a/broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionsStore.java +++ b/broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionsStore.java @@ -214,6 +214,10 @@ public class SubscriptionsStore { for (ClientTopicCouple matchingCouple : matchingSubs) { Subscription existingSub = subsForClient.get(matchingCouple.clientID); Subscription sub = m_sessionsStore.getSubscription(matchingCouple); + if (sub == null) { + //if the m_sessionStore hasn't the sub because the client disconnected + continue; + } //update the selected subscriptions if not present or if has a greater qos if (existingSub == null || existingSub.getRequestedQos().byteValue() < sub.getRequestedQos().byteValue()) { subsForClient.put(matchingCouple.clientID, sub); diff --git a/broker/src/test/java/io/moquette/server/ServerIntegrationEmbeddedPublishTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationEmbeddedPublishTest.java index 22e6a613..8d10567b 100644 --- a/broker/src/test/java/io/moquette/server/ServerIntegrationEmbeddedPublishTest.java +++ b/broker/src/test/java/io/moquette/server/ServerIntegrationEmbeddedPublishTest.java @@ -26,6 +26,7 @@ import org.junit.Test; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.io.EOFException; import java.io.IOException; import java.nio.ByteBuffer; import java.util.Properties; @@ -255,4 +256,17 @@ public class ServerIntegrationEmbeddedPublishTest { //Verify verifyMessageIsReceivedSuccessfully(); } + + @Test(expected = EOFException.class) + public void testClientSubscribeAfterDisconnected() throws Exception { + LOG.info("*** testClientSubscribeAfterDisconnected ***"); + + subscribeToWithQos("foo", QoS.AT_MOST_ONCE); + m_subscriber.disconnect(); + + internalPublishToWithQosAndRetained("foo", QOSType.MOST_ONE, false); + + //verifyMessageIsReceivedSuccessfully(); + m_subscriber.receive(2, TimeUnit.MILLISECONDS); + } }