From 7e3b6a841f6b46430ba025eef4f9cc49dc813424 Mon Sep 17 00:00:00 2001 From: andrea Date: Sun, 22 Dec 2013 17:27:43 +0100 Subject: [PATCH] Fixed issue #18, bad pubblishing of QoS0 message to inactive ( not connected) clients --- .../messaging/spi/impl/ProtocolProcessor.java | 2 +- .../spi/impl/ProtocolProcessorTest.java | 40 +++++++++++++------ .../server/ServerIntegrationFuseTest.java | 37 +++++++++++++++++ 3 files changed, 65 insertions(+), 14 deletions(-) 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 0616e493..e2822cf6 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 @@ -244,7 +244,7 @@ class ProtocolProcessor implements EventHandler { LOG.debug("Broker republishing to client <{}> topic <{}> qos <{}>, active {}", sub.getClientId(), sub.getTopic(), qos, sub.isActive()); - if (qos == AbstractMessage.QOSType.MOST_ONE) { + if (qos == AbstractMessage.QOSType.MOST_ONE && sub.isActive()) { //QoS 0 sendPublish(sub.getClientId(), topic, qos, message, false); } else { 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 16968927..3842849a 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 @@ -5,11 +5,6 @@ import java.util.Arrays; import java.util.HashMap; import java.util.List; import java.util.Map; -//import org.apache.mina.core.filterchain.IoFilter; -//import org.apache.mina.core.filterchain.IoFilterAdapter; -//import org.apache.mina.core.session.DummySession; -//import org.apache.mina.core.session.IoSession; -//import org.apache.mina.core.write.WriteRequest; import org.dna.mqtt.moquette.messaging.spi.IStorageService; import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent; import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription; @@ -22,10 +17,8 @@ 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 org.dna.mqtt.moquette.server.mina.MinaChannel; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertTrue; +import static org.junit.Assert.*; +import static org.mockito.Mockito.*; import org.junit.Before; import org.junit.Test; @@ -35,6 +28,7 @@ import org.junit.Test; */ public class ProtocolProcessorTest { final static String FAKE_CLIENT_ID = "FAKE_123"; + final static String FAKE_PUBLISHER_ID = "Publisher"; final static String FAKE_TOPIC = "/news"; ServerChannel m_session; @@ -233,7 +227,7 @@ public class ProtocolProcessorTest { }; //simulate a connect that register a clientID to an IoSession - final Subscription subscription = new Subscription(FAKE_CLIENT_ID, + final Subscription subscription = new Subscription(FAKE_PUBLISHER_ID, FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, true); //subscriptions.matches(topic) redefine the method to return true @@ -252,18 +246,18 @@ public class ProtocolProcessorTest { //simulate a connect that register a clientID to an IoSession m_processor.init(subs, m_storageService); ConnectMessage connectMessage = new ConnectMessage(); - connectMessage.setClientID(FAKE_CLIENT_ID); + connectMessage.setClientID(FAKE_PUBLISHER_ID); connectMessage.setProcotolVersion((byte)3); connectMessage.setCleanSession(subscription.isCleanSession()); m_processor.processConnect(m_session, connectMessage); ByteBuffer buffer = ByteBuffer.allocate(5).put("Hello".getBytes()); - PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, buffer, true, "FakeCLI", null); + PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, buffer, true, "Publisher", null); m_processor.processPublish(pubEvt); //Exercise SubscribeMessage msg = new SubscribeMessage(); msg.addSubscription(new SubscribeMessage.Couple((byte)QOSType.MOST_ONE.ordinal(), "#")); - m_processor.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false); + m_processor.processSubscribe(m_session, msg, FAKE_PUBLISHER_ID, false); //Verify assertNotNull(m_receivedMessage); @@ -271,4 +265,24 @@ public class ProtocolProcessorTest { PublishMessage pubMessage = (PublishMessage) m_receivedMessage; assertEquals(FAKE_TOPIC, pubMessage.getTopicName()); } + + @Test + public void processPublishNoPublishToInactiveSubscriptions() { + SubscriptionsStore mockedSubscriptions = mock(SubscriptionsStore.class); + Subscription inactiveSub = new Subscription("Subscriber", "/topic", QOSType.LEAST_ONE, false); + inactiveSub.setActive(false); + List inactiveSubscriptions = Arrays.asList(inactiveSub); + when(mockedSubscriptions.matches(eq("/topic"))).thenReturn(inactiveSubscriptions); + m_processor = new ProtocolProcessor(); + m_processor.init(mockedSubscriptions, m_storageService); + + //Exercise + ByteBuffer buffer = ByteBuffer.allocate(5).put("Hello".getBytes()); + PublishEvent pubEvt = new PublishEvent("/topic", AbstractMessage.QOSType.MOST_ONE, buffer, true, "Publisher", null); + m_processor.processPublish(pubEvt); + + //Verify no message is received + assertNull(m_receivedMessage); + } + } diff --git a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationFuseTest.java b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationFuseTest.java index 85d46c67..97c362bf 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationFuseTest.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationFuseTest.java @@ -2,6 +2,7 @@ package org.dna.mqtt.moquette.server; import java.io.File; import java.io.IOException; +import java.util.concurrent.TimeUnit; import org.fusesource.mqtt.client.BlockingConnection; import org.fusesource.mqtt.client.MQTT; import org.fusesource.mqtt.client.QoS; @@ -110,4 +111,40 @@ public class ServerIntegrationFuseTest { msg.ack(); //TODO check topic and content } + + @Test + public void checkReplayStoredPublish_forNoCleanSession_qos1() throws Exception { + LOG.info("*** checkReplayStoredPublish_forNoCleanSession_qos1 ***"); + MQTT mqtt = new MQTT(); + mqtt.setHost("localhost", 1883); + mqtt.setClientId("Publisher"); + m_publisher = mqtt.blockingConnection(); + m_publisher.connect(); + + m_mqtt.setHost("localhost", 1883); + m_mqtt.setCleanSession(false); + m_mqtt.setClientId("Subscriber"); + m_subscriber = m_mqtt.blockingConnection(); + m_subscriber.connect(); + Topic[] topics = new Topic[]{new Topic("/topic", QoS.AT_LEAST_ONCE)}; + m_subscriber.subscribe(topics); + System.out.println("***Subscriber listen to /topic and disonnect"); + m_subscriber.disconnect(); + + //force the publisher to send + m_publisher.publish("/topic", "Hello world MQTT!!-1".getBytes(), QoS.AT_LEAST_ONCE, false); + System.out.println("***Publisher published to /topic :: Hello world MQTT!!-1"); + + //reconnect and expect to receive the hello 2 message + m_subscriber = m_mqtt.blockingConnection(); + m_subscriber.connect(); + topics = new Topic[]{new Topic("/topic", QoS.AT_LEAST_ONCE)}; + m_subscriber.subscribe(topics); + Message msg = m_subscriber.receive(1, TimeUnit.SECONDS); + assertNotNull(msg); + msg.ack(); + assertEquals("Hello world MQTT!!-1", new String(msg.getPayload())); + System.out.println(String.format("***Subscriber received message <%s> from /topic", new String(msg.getPayload()))); +// m_subscriber.disconnect(); + } }