Fixed issue #1 due to bad filtering for messageType byte

This commit is contained in:
selva.andre
2012-07-24 22:43:13 +02:00
parent a17c07ad46
commit 4cd366e634
8 changed files with 31 additions and 38 deletions
@@ -18,7 +18,7 @@ public abstract class MqttDecoder extends MessageDecoderAdapter {
return NEED_DATA; return NEED_DATA;
} }
byte h1 = in.get(); byte h1 = in.get();
byte messageType = (byte) ((h1 & 0x0010) >> 4); byte messageType = (byte) ((h1 & 0x00F0) >> 4);
boolean dupFlag = ((byte) ((h1 & 0x0008) >> 3) == 1); boolean dupFlag = ((byte) ((h1 & 0x0008) >> 3) == 1);
byte qosLevel = (byte) ((h1 & 0x0006) >> 1); byte qosLevel = (byte) ((h1 & 0x0006) >> 1);
boolean retainFlag = ((byte) (h1 & 0x0001) == 1); boolean retainFlag = ((byte) (h1 & 0x0001) == 1);
@@ -1,15 +1,14 @@
package org.dna.mqtt.moquette.proto; package org.dna.mqtt.moquette.proto;
import org.dna.mqtt.moquette.proto.ConnAckDecoder;
import org.apache.mina.filter.codec.demux.MessageDecoderResult;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.apache.mina.filter.codec.demux.MessageDecoder;
import org.apache.mina.core.buffer.IoBuffer; import org.apache.mina.core.buffer.IoBuffer;
import org.apache.mina.filter.codec.demux.MessageDecoder;
import org.apache.mina.filter.codec.demux.MessageDecoderResult;
import org.dna.mqtt.moquette.proto.TestUtils.MockProtocolDecoderOutput; import org.dna.mqtt.moquette.proto.TestUtils.MockProtocolDecoderOutput;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.ConnAckMessage; import org.dna.mqtt.moquette.proto.messages.ConnAckMessage;
import static org.junit.Assert.*;
import org.junit.Before; import org.junit.Before;
import org.junit.Test; import org.junit.Test;
import static org.junit.Assert.*;
/** /**
* *
* @author andrea * @author andrea
@@ -50,6 +49,7 @@ public class ConnAckDecoderTest {
assertNotNull(m_mockProtoDecoder.getMessage()); assertNotNull(m_mockProtoDecoder.getMessage());
assertEquals(MessageDecoder.OK, res); assertEquals(MessageDecoder.OK, res);
assertEquals(ConnAckMessage.CONNECTION_ACCEPTED, m_mockProtoDecoder.getMessage().getReturnCode()); assertEquals(ConnAckMessage.CONNECTION_ACCEPTED, m_mockProtoDecoder.getMessage().getReturnCode());
assertEquals(AbstractMessage.CONNACK, m_mockProtoDecoder.getMessage().getMessageType());
} }
private void initHeader(IoBuffer buff) { private void initHeader(IoBuffer buff) {
@@ -1,18 +1,15 @@
package org.dna.mqtt.moquette.proto; package org.dna.mqtt.moquette.proto;
import org.dna.mqtt.moquette.proto.ConnectDecoder;
import org.dna.mqtt.moquette.proto.Utils;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.ConnectMessage;
import org.dna.mqtt.moquette.proto.TestUtils.MockProtocolDecoderOutput;
import org.apache.mina.filter.codec.demux.MessageDecoderResult;
import java.io.UnsupportedEncodingException; import java.io.UnsupportedEncodingException;
import org.apache.mina.core.buffer.IoBuffer; import org.apache.mina.core.buffer.IoBuffer;
import org.apache.mina.filter.codec.demux.MessageDecoder; import org.apache.mina.filter.codec.demux.MessageDecoder;
import org.apache.mina.filter.codec.demux.MessageDecoderResult;
import org.dna.mqtt.moquette.proto.TestUtils.MockProtocolDecoderOutput;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.ConnectMessage;
import static org.junit.Assert.*;
import org.junit.Before; import org.junit.Before;
import org.junit.Test; import org.junit.Test;
import static org.junit.Assert.*;
public class ConnectDecoderTest { public class ConnectDecoderTest {
@@ -121,6 +118,7 @@ public class ConnectDecoderTest {
//Verify //Verify
verifyBaseHeader(m_mockProtoDecoder.getMessage()); verifyBaseHeader(m_mockProtoDecoder.getMessage());
assertEquals(AbstractMessage.CONNECT, m_mockProtoDecoder.getMessage().getMessageType());
assertEquals("ABCDEFGH", m_mockProtoDecoder.getMessage().getClientID()); assertEquals("ABCDEFGH", m_mockProtoDecoder.getMessage().getClientID());
assertEquals("Topic", m_mockProtoDecoder.getMessage().getWillTopic()); assertEquals("Topic", m_mockProtoDecoder.getMessage().getWillTopic());
assertEquals("Message", m_mockProtoDecoder.getMessage().getWillMessage()); assertEquals("Message", m_mockProtoDecoder.getMessage().getWillMessage());
@@ -1,16 +1,14 @@
package org.dna.mqtt.moquette.proto; package org.dna.mqtt.moquette.proto;
import org.dna.mqtt.moquette.proto.PubAckDecoder;
import org.dna.mqtt.moquette.proto.Utils;
import org.apache.mina.core.buffer.IoBuffer; import org.apache.mina.core.buffer.IoBuffer;
import org.apache.mina.filter.codec.demux.MessageDecoder; import org.apache.mina.filter.codec.demux.MessageDecoder;
import org.apache.mina.filter.codec.demux.MessageDecoderResult; import org.apache.mina.filter.codec.demux.MessageDecoderResult;
import org.dna.mqtt.moquette.proto.TestUtils.MockProtocolDecoderOutput; import org.dna.mqtt.moquette.proto.TestUtils.MockProtocolDecoderOutput;
import org.dna.mqtt.moquette.proto.messages.PubAckMessage;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage; import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.PubAckMessage;
import static org.junit.Assert.*;
import org.junit.Before; import org.junit.Before;
import org.junit.Test; import org.junit.Test;
import static org.junit.Assert.*;
/** /**
* *
@@ -41,6 +39,7 @@ public class PubAckDecoderTest {
assertNotNull(m_mockProtoDecoder.getMessage()); assertNotNull(m_mockProtoDecoder.getMessage());
assertEquals(MessageDecoder.OK, res); assertEquals(MessageDecoder.OK, res);
assertEquals(messageId, m_mockProtoDecoder.getMessage().getMessageID()); assertEquals(messageId, m_mockProtoDecoder.getMessage().getMessageID());
assertEquals(AbstractMessage.PUBACK, m_mockProtoDecoder.getMessage().getMessageType());
} }
private void initHeader(IoBuffer buff, int messageID) { private void initHeader(IoBuffer buff, int messageID) {
@@ -1,16 +1,14 @@
package org.dna.mqtt.moquette.proto; package org.dna.mqtt.moquette.proto;
import org.dna.mqtt.moquette.proto.PublishDecoder;
import org.dna.mqtt.moquette.proto.Utils;
import org.apache.mina.filter.codec.demux.MessageDecoderResult;
import org.dna.mqtt.moquette.proto.messages.PublishMessage;
import org.apache.mina.core.buffer.IoBuffer; import org.apache.mina.core.buffer.IoBuffer;
import org.apache.mina.filter.codec.demux.MessageDecoder; import org.apache.mina.filter.codec.demux.MessageDecoder;
import org.apache.mina.filter.codec.demux.MessageDecoderResult;
import org.dna.mqtt.moquette.proto.TestUtils.MockProtocolDecoderOutput; import org.dna.mqtt.moquette.proto.TestUtils.MockProtocolDecoderOutput;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage; import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.PublishMessage;
import static org.junit.Assert.*;
import org.junit.Before; import org.junit.Before;
import org.junit.Test; import org.junit.Test;
import static org.junit.Assert.*;
/** /**
* *
@@ -48,6 +46,7 @@ public class PublishDecoderTest {
assertEquals(MessageDecoder.OK, res); assertEquals(MessageDecoder.OK, res);
assertEquals("Fake Topic", m_mockProtoDecoder.getMessage().getTopicName()); assertEquals("Fake Topic", m_mockProtoDecoder.getMessage().getTopicName());
assertNull(m_mockProtoDecoder.getMessage().getMessageID()); assertNull(m_mockProtoDecoder.getMessage().getMessageID());
assertEquals(AbstractMessage.PUBLISH, m_mockProtoDecoder.getMessage().getMessageType());
} }
@Test @Test
@@ -1,17 +1,15 @@
package org.dna.mqtt.moquette.proto; package org.dna.mqtt.moquette.proto;
import org.dna.mqtt.moquette.proto.SubAckDecoder;
import org.dna.mqtt.moquette.proto.Utils;
import java.util.List; import java.util.List;
import org.dna.mqtt.moquette.proto.messages.SubAckMessage;
import org.apache.mina.core.buffer.IoBuffer; import org.apache.mina.core.buffer.IoBuffer;
import org.apache.mina.filter.codec.demux.MessageDecoderResult; import org.apache.mina.filter.codec.demux.MessageDecoderResult;
import org.dna.mqtt.moquette.proto.TestUtils.MockProtocolDecoderOutput; import org.dna.mqtt.moquette.proto.TestUtils.MockProtocolDecoderOutput;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage; import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
import org.dna.mqtt.moquette.proto.messages.SubAckMessage;
import static org.junit.Assert.*;
import org.junit.Before; import org.junit.Before;
import org.junit.Test; import org.junit.Test;
import static org.junit.Assert.*;
/** /**
* *
@@ -46,10 +44,11 @@ public class SubAckDecoderTest {
assertEquals(QOSType.LEAST_ONE, qoses.get(0)); assertEquals(QOSType.LEAST_ONE, qoses.get(0));
assertEquals(QOSType.MOST_ONE, qoses.get(1)); assertEquals(QOSType.MOST_ONE, qoses.get(1));
assertEquals(QOSType.MOST_ONE, qoses.get(2)); assertEquals(QOSType.MOST_ONE, qoses.get(2));
assertEquals(AbstractMessage.SUBACK, m_mockProtoDecoder.getMessage().getMessageType());
} }
private void initHeaderQos(IoBuffer buff, int messageID, QOSType... qoss) throws IllegalAccessException { private void initHeaderQos(IoBuffer buff, int messageID, QOSType... qoss) throws IllegalAccessException {
buff.clear().put((byte) (AbstractMessage.SUBSCRIBE << 4)). buff.clear().put((byte) (AbstractMessage.SUBACK << 4)).
put(Utils.encodeRemainingLength(3)); put(Utils.encodeRemainingLength(3));
Utils.writeWord(buff, messageID); Utils.writeWord(buff, messageID);
@@ -1,16 +1,14 @@
package org.dna.mqtt.moquette.proto; package org.dna.mqtt.moquette.proto;
import org.dna.mqtt.moquette.proto.SubscribeDecoder;
import org.dna.mqtt.moquette.proto.Utils;
import org.dna.mqtt.moquette.proto.messages.SubscribeMessage.Couple;
import org.dna.mqtt.moquette.proto.messages.SubscribeMessage;
import org.apache.mina.core.buffer.IoBuffer; import org.apache.mina.core.buffer.IoBuffer;
import org.apache.mina.filter.codec.demux.MessageDecoderResult; import org.apache.mina.filter.codec.demux.MessageDecoderResult;
import org.dna.mqtt.moquette.proto.TestUtils.MockProtocolDecoderOutput; import org.dna.mqtt.moquette.proto.TestUtils.MockProtocolDecoderOutput;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage; import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.SubscribeMessage;
import org.dna.mqtt.moquette.proto.messages.SubscribeMessage.Couple;
import static org.junit.Assert.*;
import org.junit.Before; import org.junit.Before;
import org.junit.Test; import org.junit.Test;
import static org.junit.Assert.*;
/** /**
* *
@@ -56,6 +54,7 @@ public class SubscribeDecoderTest {
//Verify //Verify
assertEquals(MessageDecoderResult.OK, res); assertEquals(MessageDecoderResult.OK, res);
assertEquals(2, m_mockProtoDecoder.getMessage().subscriptions().size()); assertEquals(2, m_mockProtoDecoder.getMessage().subscriptions().size());
assertEquals(AbstractMessage.SUBSCRIBE, m_mockProtoDecoder.getMessage().getMessageType());
} }
private void initHeaderBadQos(IoBuffer buff) { private void initHeaderBadQos(IoBuffer buff) {
@@ -1,15 +1,13 @@
package org.dna.mqtt.moquette.proto; package org.dna.mqtt.moquette.proto;
import org.dna.mqtt.moquette.proto.Utils;
import org.dna.mqtt.moquette.proto.UnsubscribeDecoder;
import org.dna.mqtt.moquette.proto.messages.UnsubscribeMessage;
import org.apache.mina.core.buffer.IoBuffer; import org.apache.mina.core.buffer.IoBuffer;
import org.apache.mina.filter.codec.demux.MessageDecoderResult; import org.apache.mina.filter.codec.demux.MessageDecoderResult;
import org.dna.mqtt.moquette.proto.TestUtils.MockProtocolDecoderOutput; import org.dna.mqtt.moquette.proto.TestUtils.MockProtocolDecoderOutput;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage; import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.UnsubscribeMessage;
import static org.junit.Assert.*;
import org.junit.Before; import org.junit.Before;
import org.junit.Test; import org.junit.Test;
import static org.junit.Assert.*;
/** /**
* *
@@ -57,6 +55,7 @@ public class UnsubscribeDecoderTest {
assertEquals(2, m_mockProtoDecoder.getMessage().topics().size()); assertEquals(2, m_mockProtoDecoder.getMessage().topics().size());
assertEquals(topic1, m_mockProtoDecoder.getMessage().topics().get(0)); assertEquals(topic1, m_mockProtoDecoder.getMessage().topics().get(0));
assertEquals(topic2, m_mockProtoDecoder.getMessage().topics().get(1)); assertEquals(topic2, m_mockProtoDecoder.getMessage().topics().get(1));
assertEquals(AbstractMessage.UNSUBSCRIBE, m_mockProtoDecoder.getMessage().getMessageType());
} }
private void initHeaderBadQos(IoBuffer buff) { private void initHeaderBadQos(IoBuffer buff) {
@@ -71,7 +70,7 @@ public class UnsubscribeDecoderTest {
} }
topicBuffer.flip(); topicBuffer.flip();
buff.clear().put((byte)(AbstractMessage.SUBSCRIBE << 4 | (byte)0x02)) buff.clear().put((byte)(AbstractMessage.UNSUBSCRIBE << 4 | (byte)0x02))
.put(Utils.encodeRemainingLength(topicBuffer.remaining())); .put(Utils.encodeRemainingLength(topicBuffer.remaining()));
buff.put(topicBuffer); buff.put(topicBuffer);
} }