From afcc20f483ee50a2c52eab1cf7b275ce45dd14d1 Mon Sep 17 00:00:00 2001 From: andrea Date: Wed, 28 Aug 2013 10:26:02 +0200 Subject: [PATCH] Added unit test and fix for issue #12, due to bad topic in publish message after a late subscription and messages retained in DB. --- .../spi/impl/HawtDBStorageService.java | 10 ++++- .../messaging/spi/impl/SimpleMessaging.java | 4 +- .../spi/impl/SimpleMessagingTest.java | 44 +++++++++++++++++++ 3 files changed, 54 insertions(+), 4 deletions(-) diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java index 88e1386e..a89edea7 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java @@ -26,10 +26,12 @@ public class HawtDBStorageService implements IStorageService { public static class StoredMessage implements Serializable { AbstractMessage.QOSType m_qos; byte[] m_payload; + String m_topic; - StoredMessage(byte[] message, AbstractMessage.QOSType qos) { + StoredMessage(byte[] message, AbstractMessage.QOSType qos, String topic) { m_qos = qos; m_payload = message; + m_topic = topic; } AbstractMessage.QOSType getQos() { @@ -39,6 +41,10 @@ public class HawtDBStorageService implements IStorageService { byte[] getPayload() { return m_payload; } + + String getTopic() { + return m_topic; + } } @@ -132,7 +138,7 @@ public class HawtDBStorageService implements IStorageService { m_retainedStore.remove(topic); } else { //store the message to the topic - m_retainedStore.put(topic, new StoredMessage(message, qos)); + m_retainedStore.put(topic, new StoredMessage(message, qos, topic)); } } diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java index db4d0085..a629243d 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java @@ -400,7 +400,7 @@ public class SimpleMessaging implements IMessaging, EventHandler { for (StoredMessage storedMsg : messages) { //fire the as retained the message LOG.debug("Inserting NotifyEvent into outbound for topic " + topic); - notify(new NotifyEvent(newSubscription.getClientId(), topic, storedMsg.getQos(), storedMsg.getPayload(), true)); + notify(new NotifyEvent(newSubscription.getClientId(), storedMsg.getTopic(), storedMsg.getQos(), storedMsg.getPayload(), true)); } } @@ -408,7 +408,7 @@ public class SimpleMessaging implements IMessaging, EventHandler { LOG.debug("processSubscribe invoked"); for (SubscribeMessage.Couple req : msg.subscriptions()) { - QOSType qos = AbstractMessage.QOSType.fromByte(req.getQos()); + QOSType qos = AbstractMessage.QOSType.values()[req.getQos()]; Subscription newSubscription = new Subscription(clientID, req.getTopic(), qos, cleanSession); subscribeSingleTopic(newSubscription, req.getTopic()); } diff --git a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java index 31f0ad90..826488d7 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java @@ -1,5 +1,6 @@ package org.dna.mqtt.moquette.messaging.spi.impl; +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; @@ -124,6 +125,49 @@ public class SimpleMessagingTest { //TODO check received message attributes } + @Test + public void testPublishOfRetainedMessage() throws Exception { + m_session.getFilterChain().remove("MessageCatcher"); + m_session.getFilterChain().addFirst("MessageCatcher", new IoFilterAdapter() { + + @Override + public void filterWrite(IoFilter.NextFilter nextFilter, IoSession session, + WriteRequest writeRequest) throws Exception { + try { + System.out.println("filterReceived class " + writeRequest.getMessage().getClass().getName()); + if (writeRequest.getMessage() instanceof PublishMessage) { + m_receivedMessage = (AbstractMessage) writeRequest.getMessage(); + } + + if (m_receivedMessage instanceof ConnAckMessage) { + ConnAckMessage buf = (ConnAckMessage) m_receivedMessage; + m_returnCode = buf.getReturnCode(); + } + } catch (Exception ex) { + throw new AssertionError("Wrong return code"); + } + } + }); + + //simulate a connect that register a clientID to an IoSession + ConnectionDescriptor connDescr = new ConnectionDescriptor(FAKE_CLIENT_ID, m_session, true); + messaging.m_clientIDs.put(FAKE_CLIENT_ID, connDescr); + + PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, QOSType.MOST_ONE, "Hello".getBytes(), true, "FakeCLI", null); + messaging.processPublish(pubEvt); + + //Exercise + SubscribeMessage msg = new SubscribeMessage(); + msg.addSubscription(new SubscribeMessage.Couple((byte)QOSType.MOST_ONE.ordinal(), "#")); + messaging.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false); + + //Verify + assertNotNull(m_receivedMessage); + assertTrue(m_receivedMessage instanceof PublishMessage); + PublishMessage pubMessage = (PublishMessage) m_receivedMessage; + assertEquals(FAKE_TOPIC, pubMessage.getTopicName()); + } + @Test public void testHandleConnect_BadProtocol() { connMsg.setProcotolVersion((byte) 0x02);