anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Updated publish message parto to conform to 3.1.1 specification

Browse Source
This commit is contained in:
selva.andre
2014-07-31 00:08:41 +02:00
parent 482ffe0c65
commit d4ef81a7ef
9 changed files with 160 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessagesStore.java
+2
View File
@@ -68,4 +68,6 @@ public interface IMessagesStore {
void removeQoS2Message(String publishKey);
PublishEvent retrieveQoS2Message(String publishKey);
void cleanRetained(String topic);
}
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBPersistentStore.java
+5
View File
@@ -150,6 +150,11 @@ public class HawtDBPersistentStore implements IMessagesStore, ISessionsStore {
m_qos2Store = (SortedIndex<String, StoredPublishEvent>) 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()) {
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java
+20 -2
View File
@@ -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<HawtDBPersistentStore.StoredMessage> 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<PublishEvent> 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<PublishEvent> 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<Subscription>());
@@ -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<Subscription> listAllSubscriptions() {
List<Subscription> allSubscriptions = new ArrayList<Subscription>();
for (Map.Entry<String, Set<Subscription>> 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);
}
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java
+7 -2
View File
@@ -303,7 +303,12 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
}
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<ValueEvent> {
}
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);
broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java
+30
View File
@@ -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<PublishEvent> retainedPublishes = m_storageService.retrievePersistedPublishes("Publisher");
Collection<HawtDBPersistentStore.StoredMessage> messages = m_storageService.searchMatching(new IMatchingCondition() {
public boolean match(String key) {
return SubscriptionsStore.matchTopics(key, FAKE_TOPIC);
}
});
assertTrue(messages.isEmpty());
}
}
broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationPahoTest.java
+1 -1
View File
@@ -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();
netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PublishDecoder.java
+23
View File
@@ -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<Object> out) throws Exception {
decode((AttributeMap)ctx, in, out);
}
// @Override
void decode(AttributeMap ctx, ByteBuf in, List<Object> 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 ||
netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/Utils.java
+11
View File
@@ -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<Integer> versionAttr = attrsMap.attr(MQTTDecoder.PROTOCOL_VERSION);
Integer protocolVersion = versionAttr.get();
if (protocolVersion == null) {
return true;
}
return protocolVersion == VERSION_3_1_1;
}
}
netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/PublishDecoderTest.java
+61 -13
View File
@@ -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<Object> m_results;
AttributeMap m_attrMap;
private static final int MESSAGE_ID = 123;
@Before
public void setUp() {
m_msgdec = new PublishDecoder();
m_results = new ArrayList<Object >();
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;
}
}