From d4ef81a7ef7335f6f2dabe47f37ef4198ccd0a3e Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Thu, 31 Jul 2014 00:08:41 +0200 Subject: [PATCH] Updated publish message parto to conform to 3.1.1 specification --- .../messaging/spi/IMessagesStore.java | 2 + .../spi/impl/HawtDBPersistentStore.java | 5 ++ .../spi/impl/MemoryStorageService.java | 22 +++++- .../messaging/spi/impl/ProtocolProcessor.java | 9 ++- .../spi/impl/ProtocolProcessorTest.java | 30 ++++++++ .../server/ServerIntegrationPahoTest.java | 2 +- .../moquette/parser/netty/PublishDecoder.java | 23 ++++++ .../dna/mqtt/moquette/parser/netty/Utils.java | 11 +++ .../parser/netty/PublishDecoderTest.java | 74 +++++++++++++++---- 9 files changed, 160 insertions(+), 18 deletions(-) diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessagesStore.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessagesStore.java index d3fafe5a..d21b9918 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessagesStore.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessagesStore.java @@ -68,4 +68,6 @@ public interface IMessagesStore { void removeQoS2Message(String publishKey); PublishEvent retrieveQoS2Message(String publishKey); + + void cleanRetained(String topic); } diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBPersistentStore.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBPersistentStore.java index fd6478a5..e53aad8f 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBPersistentStore.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBPersistentStore.java @@ -150,6 +150,11 @@ public class HawtDBPersistentStore implements IMessagesStore, ISessionsStore { m_qos2Store = (SortedIndex) m_multiIndexFactory.openOrCreate("qos2Store", indexFactory); } + @Override + public void cleanRetained(String topic) { + m_retainedStore.remove(topic); + } + @Override public void storeRetained(String topic, ByteBuffer message, AbstractMessage.QOSType qos) { if (!message.hasRemaining()) { diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java index 61dd8531..cabc1b2c 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java @@ -47,9 +47,14 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore { private static final Logger LOG = LoggerFactory.getLogger(MemoryStorageService.class); public void initStore() { - //To change body of implemented methods use File | Settings | File Templates. } - + + @Override + public void cleanRetained(String topic) { + m_retainedStore.remove(topic); + } + + @Override public void storeRetained(String topic, ByteBuffer message, AbstractMessage.QOSType qos) { if (!message.hasRemaining()) { //clean the message from topic @@ -62,6 +67,7 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore { } } + @Override public Collection searchMatching(IMatchingCondition condition) { LOG.debug("searchMatching scanning all retained messages, presents are {}", m_retainedStore.size()); @@ -77,6 +83,7 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore { return results; } + @Override public void storePublishForFuture(PublishEvent evt) { LOG.debug("storePublishForFuture store evt {}", evt); List storedEvents; @@ -95,6 +102,7 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore { return m_persistentMessageStore.get(clientID); } + @Override public void cleanPersistedPublishMessage(String clientID, int messageID) { List events = m_persistentMessageStore.get(clientID); PublishEvent toRemoveEvt = null; @@ -107,18 +115,22 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore { m_persistentMessageStore.put(clientID, events); } + @Override public void cleanPersistedPublishes(String clientID) { m_persistentMessageStore.remove(clientID); } + @Override public void cleanInFlight(String msgID) { m_inflightStore.remove(msgID); } + @Override public void addInFlight(PublishEvent evt, String publishKey) { m_inflightStore.put(publishKey, evt); } + @Override public void addNewSubscription(Subscription newSubscription, String clientID) { if (!m_persistentSubscriptions.containsKey(clientID)) { m_persistentSubscriptions.put(clientID, new HashSet()); @@ -131,6 +143,7 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore { } } + @Override public void wipeSubscriptions(String clientID) { m_persistentSubscriptions.remove(clientID); } @@ -140,6 +153,7 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore { return m_persistentSubscriptions.containsKey(clientID); } + @Override public List listAllSubscriptions() { List allSubscriptions = new ArrayList(); for (Map.Entry> entry : m_persistentSubscriptions.entrySet()) { @@ -148,19 +162,23 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore { return allSubscriptions; } + @Override public void close() { //To change body of implemented methods use File | Settings | File Templates. } + @Override public void persistQoS2Message(String publishKey, PublishEvent evt) { LOG.debug("persistQoS2Message store pubKey {}, evt {}", publishKey, evt); m_qos2Store.put(publishKey, evt); } + @Override public void removeQoS2Message(String publishKey) { m_qos2Store.remove(publishKey); } + @Override public PublishEvent retrieveQoS2Message(String publishKey) { return m_qos2Store.get(publishKey); } 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 537be8c4..1a4c6535 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 @@ -303,7 +303,12 @@ class ProtocolProcessor implements EventHandler { } if (retain) { - m_messagesStore.storeRetained(topic, message, qos); + if (qos == AbstractMessage.QOSType.MOST_ONE) { + //QoS == 0 && retain => clean old retained + m_messagesStore.cleanRetained(topic); + } else { + m_messagesStore.storeRetained(topic, message, qos); + } } } @@ -408,7 +413,7 @@ class ProtocolProcessor implements EventHandler { } LOG.debug("clientIDs are {}", m_clientIDs); if (m_clientIDs.get(clientId) == null) { - throw new RuntimeException(String.format("Can't find a ConnectionDEwcriptor for client %s in cache %s", clientId, m_clientIDs)); + throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client %s in cache %s", clientId, m_clientIDs)); } // LOG.debug("Session for clientId " + clientId + " is " + m_clientIDs.get(clientId).getSession()); // m_clientIDs.get(clientId).getSession().write(pubAckMessage); 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 050aa9d4..c7c6f635 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 @@ -17,12 +17,14 @@ package org.dna.mqtt.moquette.messaging.spi.impl; import java.nio.ByteBuffer; import java.util.Arrays; +import java.util.Collection; import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import org.dna.mqtt.moquette.messaging.spi.IMatchingCondition; import org.dna.mqtt.moquette.messaging.spi.IMessagesStore; import org.dna.mqtt.moquette.messaging.spi.ISessionsStore; import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent; @@ -41,6 +43,7 @@ import org.dna.mqtt.moquette.server.ServerChannel; import static org.junit.Assert.*; import org.junit.Before; import org.junit.Test; +import static org.junit.internal.runners.model.MultipleFailureException.assertEmpty; import static org.mockito.Mockito.*; /** @@ -506,4 +509,31 @@ public class ProtocolProcessorTest { //Verify no message is received assertNull(m_receivedMessage); } + + + /** + * Verify that receiving a publish with retained message and with Q0S = 0 + * clean the existing retained messages for that topic. + */ + @Test + public void testCleanRetainedStoreAfterAQoS0AndRetainedTrue() { + //prepare and existing retained store + ByteBuffer payload = ByteBuffer.allocate(5).put("Hello".getBytes()); + PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, AbstractMessage.QOSType.LEAST_ONE, payload, true, "Publisher"); + m_processor.processPublish(pubEvt); + + //Exercise + PublishEvent cleanPubEvt = new PublishEvent(FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, payload, true, "Publisher"); + m_processor.processPublish(cleanPubEvt); + + //Verify +// List retainedPublishes = m_storageService.retrievePersistedPublishes("Publisher"); + + Collection messages = m_storageService.searchMatching(new IMatchingCondition() { + public boolean match(String key) { + return SubscriptionsStore.matchTopics(key, FAKE_TOPIC); + } + }); + assertTrue(messages.isEmpty()); + } } diff --git a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationPahoTest.java b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationPahoTest.java index 2690abb7..d076d096 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationPahoTest.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationPahoTest.java @@ -172,7 +172,7 @@ public class ServerIntegrationPahoTest { public void testRetain_maintainMessage_againstClientDestruction() throws Exception { LOG.info("*** testRetain_maintainMessage_againstClientDestruction ***"); m_client.connect(); - m_client.publish("/topic", "Test my payload".getBytes(), 0, true); + m_client.publish("/topic", "Test my payload".getBytes(), 1, true); m_client.disconnect(); diff --git a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PublishDecoder.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PublishDecoder.java index a610dba8..88afa29c 100644 --- a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PublishDecoder.java +++ b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PublishDecoder.java @@ -18,6 +18,8 @@ package org.dna.mqtt.moquette.parser.netty; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.ChannelHandlerContext; +import io.netty.handler.codec.CorruptedFrameException; +import io.netty.util.AttributeMap; import java.util.List; import org.dna.mqtt.moquette.proto.messages.AbstractMessage; import org.dna.mqtt.moquette.proto.messages.PublishMessage; @@ -34,6 +36,11 @@ class PublishDecoder extends DemuxDecoder { @Override void decode(ChannelHandlerContext ctx, ByteBuf in, List out) throws Exception { + decode((AttributeMap)ctx, in, out); + } + +// @Override + void decode(AttributeMap ctx, ByteBuf in, List out) throws Exception { LOG.info("decode invoked with buffer {}", in); in.resetReaderIndex(); int startPos = in.readerIndex(); @@ -45,6 +52,18 @@ class PublishDecoder extends DemuxDecoder { in.resetReaderIndex(); return; } + + if (Utils.isMQTT3_1_1(ctx)) { + if (message.getQos() == AbstractMessage.QOSType.MOST_ONE && message.isDupFlag()) { + //bad protocol, if QoS=0 => DUP = 0 + throw new CorruptedFrameException("Received a PUBLISH with QoS=0 & DUP = 1, MQTT 3.1.1 violation"); + } + + if (message.getQos() == AbstractMessage.QOSType.RESERVED) { + throw new CorruptedFrameException("Received a PUBLISH with QoS flags setted 10 b11, MQTT 3.1.1 violation"); + } + } + int remainingLength = message.getRemainingLength(); //Topic name @@ -53,6 +72,10 @@ class PublishDecoder extends DemuxDecoder { in.resetReaderIndex(); return; } + if (topic.contains("+") || topic.contains("#")) { + throw new CorruptedFrameException("Received a PUBLISH with topic containting wild card chars, topic: " + topic); + } + message.setTopicName(topic); if (message.getQos() == AbstractMessage.QOSType.LEAST_ONE || diff --git a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/Utils.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/Utils.java index ae34484b..684312b1 100644 --- a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/Utils.java +++ b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/Utils.java @@ -18,6 +18,8 @@ package org.dna.mqtt.moquette.parser.netty; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.handler.codec.CorruptedFrameException; +import io.netty.util.Attribute; +import io.netty.util.AttributeMap; import java.io.UnsupportedEncodingException; import org.dna.mqtt.moquette.proto.messages.AbstractMessage; import org.slf4j.LoggerFactory; @@ -174,4 +176,13 @@ public class Utils { flags |= ((message.getQos().ordinal() & 0x03) << 1); return flags; } + + static boolean isMQTT3_1_1(AttributeMap attrsMap) { + Attribute versionAttr = attrsMap.attr(MQTTDecoder.PROTOCOL_VERSION); + Integer protocolVersion = versionAttr.get(); + if (protocolVersion == null) { + return true; + } + return protocolVersion == VERSION_3_1_1; + } } diff --git a/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/PublishDecoderTest.java b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/PublishDecoderTest.java index c34aa510..cda125fe 100644 --- a/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/PublishDecoderTest.java +++ b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/PublishDecoderTest.java @@ -17,6 +17,9 @@ package org.dna.mqtt.moquette.parser.netty; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; +import io.netty.handler.codec.CorruptedFrameException; +import io.netty.util.AttributeMap; +import io.netty.util.DefaultAttributeMap; import java.nio.Buffer; import java.nio.ByteBuffer; import java.util.ArrayList; @@ -38,11 +41,15 @@ public class PublishDecoderTest { ByteBuf m_buff; PublishDecoder m_msgdec; List m_results; + AttributeMap m_attrMap; + + private static final int MESSAGE_ID = 123; @Before public void setUp() { m_msgdec = new PublishDecoder(); m_results = new ArrayList(); + m_attrMap = new DefaultAttributeMap(); } @Test @@ -51,7 +58,7 @@ public class PublishDecoderTest { initHeader(m_buff); //Excercise - m_msgdec.decode(null, m_buff, m_results); + m_msgdec.decode(m_attrMap, m_buff, m_results); assertFalse(m_results.isEmpty()); PublishMessage message = (PublishMessage)m_results.get(0); @@ -64,40 +71,67 @@ public class PublishDecoderTest { @Test public void testHeaderWithMessageID() throws Exception { m_buff = Unpooled.buffer(14); - int messageID = 123; - initHeaderWithMessageID(m_buff, messageID); + initHeaderWithMessageID(m_buff, MESSAGE_ID); //Exercise - m_msgdec.decode(null, m_buff, m_results); + m_msgdec.decode(m_attrMap, m_buff, m_results); assertFalse(m_results.isEmpty()); PublishMessage message = (PublishMessage)m_results.get(0); assertNotNull(message); assertEquals("Fake Topic", message.getTopicName()); - assertEquals(messageID, (int) message.getMessageID()); + assertEquals(MESSAGE_ID, (int) message.getMessageID()); } @Test public void testHeaderWithMessageID_Payload() throws Exception { m_buff = Unpooled.buffer(14); - int messageID = 123; // byte[] payload = new byte[]{0x0A, 0x0B, 0x0C}; ByteBuffer payload = ByteBuffer.allocate(3).put(new byte[]{0x0A, 0x0B, 0x0C}); - initHeaderWithMessageID_Payload(m_buff, messageID, payload); + initHeaderWithMessageID_Payload(m_buff, MESSAGE_ID, payload); //Exercise - m_msgdec.decode(null, m_buff, m_results); + m_msgdec.decode(m_attrMap, m_buff, m_results); assertFalse(m_results.isEmpty()); PublishMessage message = (PublishMessage)m_results.get(0); assertNotNull(message); assertEquals("Fake Topic", message.getTopicName()); - assertEquals(messageID, (int) message.getMessageID()); + assertEquals(MESSAGE_ID, (int) message.getMessageID()); // TestUtils.verifyEquals(payload, message.getPayload()); assertEquals(payload, message.getPayload()); } + @Test(expected = CorruptedFrameException.class) + public void testDup0WithQoS0_3_1_1() throws Exception { + m_buff = m_buff = preparePubclishWithQosFlags((byte) 0x08); + + //Exercise + m_msgdec.decode(m_attrMap, m_buff, m_results); + } + + @Test(expected = CorruptedFrameException.class) + public void testReservedQoS3_3_1_1() throws Exception { + m_buff = preparePubclishWithQosFlags((byte) 0x0E); + + //Exercise + m_msgdec.decode(m_attrMap, m_buff, m_results); + } + + @Test(expected = CorruptedFrameException.class) + public void testTopicWithWildCards() throws Exception { + byte[] overallMessage = new byte[]{0x30, 0x17, //fixed header, 25 byte lenght + 0x00, 0x06, 0x2f, 0x74, 0x6f, 0x70, 0x2B /*+*/, 0x23 /*#*/, //[/top+#] string 2 len + 6 content + 0x54, 0x65, 0x73, 0x74, 0x20, 0x6d, 0x79, // [Test my payload] encoding + 0x20, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64}; + m_buff = Unpooled.buffer(overallMessage.length); + m_buff.writeBytes(overallMessage); + + //Exercise + m_msgdec.decode(m_attrMap, m_buff, m_results); + } + @Test public void testBugOnRealCase() throws Exception { byte[] overallMessage = new byte[]{0x30, 0x17, //fixed header, 25 byte lenght @@ -108,7 +142,7 @@ public class PublishDecoderTest { m_buff.writeBytes(overallMessage); //Exercise - m_msgdec.decode(null, m_buff, m_results); + m_msgdec.decode(m_attrMap, m_buff, m_results); assertFalse(m_results.isEmpty()); PublishMessage message = (PublishMessage)m_results.get(0); @@ -128,14 +162,14 @@ public class PublishDecoderTest { //Exercise - m_msgdec.decode(null, doubleMessageBuf, m_results); + m_msgdec.decode(m_attrMap, doubleMessageBuf, m_results); assertFalse(m_results.isEmpty()); PublishMessage message = (PublishMessage)m_results.get(0); assertNotNull(message); m_results.clear(); - m_msgdec.decode(null, doubleMessageBuf, m_results); + m_msgdec.decode(m_attrMap, doubleMessageBuf, m_results); assertFalse(m_results.isEmpty()); message = (PublishMessage)m_results.get(0); @@ -154,7 +188,7 @@ public class PublishDecoderTest { msgBuf.readByte(); //to simulate the reading of messageType done by MQTTDecoder dispatcher //Exercise - m_msgdec.decode(null, msgBuf, m_results); + m_msgdec.decode(m_attrMap, msgBuf, m_results); assertFalse(m_results.isEmpty()); PublishMessage message = (PublishMessage)m_results.get(0); @@ -216,4 +250,18 @@ public class PublishDecoderTest { return completeMsg; } + + private ByteBuf preparePubclishWithQosFlags(byte flags) { + ByteBuf buff = Unpooled.buffer(14); + ByteBuffer payload = ByteBuffer.allocate(3).put(new byte[]{0x0A, 0x0B, 0x0C}); + ByteBuf tmp = Unpooled.buffer(4).writeBytes(Utils.encodeString("Fake Topic")); + tmp.writeShort(MESSAGE_ID); + tmp.writeBytes(payload); + buff.clear().writeByte(AbstractMessage.PUBLISH << 4 | flags) //set DUP=1 Qos to 11 => b1110 + .writeBytes(Utils.encodeRemainingLength(tmp.readableBytes())); + //topic name + buff.writeBytes(tmp); + + return buff; + } }