From ecea97f4098f8fa40da35b0b06983cc0b90038cd Mon Sep 17 00:00:00 2001 From: andrea Date: Sun, 16 Sep 2012 17:22:13 +0200 Subject: [PATCH] Fixed issue 4, bad calculation of remaining length in big PublishDecoder. --- .../mqtt/moquette/proto/PublishDecoder.java | 120 +++--- .../mqtt/moquette/proto/PublishEncoder.java | 96 ++--- .../org/dna/mqtt/moquette/proto/Utils.java | 371 +++++++++--------- .../moquette/proto/PublishDecoderTest.java | 312 ++++++++------- .../moquette/proto/PublishEncoderTest.java | 312 ++++++++------- .../dna/mqtt/moquette/proto/TestUtils.java | 221 ++++++----- 6 files changed, 773 insertions(+), 659 deletions(-) diff --git a/parser/src/main/java/org/dna/mqtt/moquette/proto/PublishDecoder.java b/parser/src/main/java/org/dna/mqtt/moquette/proto/PublishDecoder.java index f98aa81d..e879e09f 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/proto/PublishDecoder.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/proto/PublishDecoder.java @@ -1,57 +1,63 @@ -package org.dna.mqtt.moquette.proto; - -import org.apache.mina.core.buffer.IoBuffer; -import org.apache.mina.core.session.IoSession; -import org.apache.mina.filter.codec.ProtocolDecoderOutput; -import org.apache.mina.filter.codec.demux.MessageDecoderResult; -import org.dna.mqtt.moquette.proto.messages.AbstractMessage; -import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; -import org.dna.mqtt.moquette.proto.messages.PublishMessage; - -/** - * - * @author andrea - */ -public class PublishDecoder extends MqttDecoder { - - public MessageDecoderResult decodable(IoSession session, IoBuffer in) { - return Utils.checkDecodable(AbstractMessage.PUBLISH, in); - } - - public MessageDecoderResult decode(IoSession session, IoBuffer in, ProtocolDecoderOutput out) throws Exception { - int startPos = in.position(); - - //Common decoding part - PublishMessage message = new PublishMessage(); - if (decodeCommonHeader(message, in) == NEED_DATA) { - return NEED_DATA; - } - int remainingLength = message.getRemainingLength(); - - //Topic name - String topic = Utils.decodeString(in); - if (topic == null) { - return NEED_DATA; - } - message.setTopicName(topic); - - if (message.getQos() == QOSType.LEAST_ONE || - message.getQos() == QOSType.EXACTLY_ONCE) { - message.setMessageID(Utils.readWord(in)); - } - int stopPos = in.position(); - - //read the payload - int payloadSize = remainingLength - (stopPos - startPos - 2); - byte[] b = new byte[payloadSize]; - if (in.remaining() < payloadSize) { - return NEED_DATA; - } - in.get(b); - message.setPayload(b); - - out.write(message); - return OK; - } - -} +package org.dna.mqtt.moquette.proto; + +import org.apache.mina.core.buffer.IoBuffer; +import org.apache.mina.core.session.IoSession; +import org.apache.mina.filter.codec.ProtocolDecoderOutput; +import org.apache.mina.filter.codec.demux.MessageDecoderResult; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; +import org.dna.mqtt.moquette.proto.messages.PublishMessage; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * + * @author andrea + */ +public class PublishDecoder extends MqttDecoder { + + private static Logger LOG = LoggerFactory.getLogger(PublishDecoder.class); + + public MessageDecoderResult decodable(IoSession session, IoBuffer in) { + return Utils.checkDecodable(AbstractMessage.PUBLISH, in); + } + + public MessageDecoderResult decode(IoSession session, IoBuffer in, ProtocolDecoderOutput out) throws Exception { + LOG.debug("decode invoked with buffer " + in); + int startPos = in.position(); + + //Common decoding part + PublishMessage message = new PublishMessage(); + if (decodeCommonHeader(message, in) == NEED_DATA) { + LOG.info("decode ask for more data after " + in); + return NEED_DATA; + } + int remainingLength = message.getRemainingLength(); + + //Topic name + String topic = Utils.decodeString(in); + if (topic == null) { + return NEED_DATA; + } + message.setTopicName(topic); + + if (message.getQos() == QOSType.LEAST_ONE || + message.getQos() == QOSType.EXACTLY_ONCE) { + message.setMessageID(Utils.readWord(in)); + } + int stopPos = in.position(); + + //read the payload + int payloadSize = remainingLength - (stopPos - startPos - 2) + (Utils.numBytesToEncode(remainingLength) - 1); + if (in.remaining() < payloadSize) { + return NEED_DATA; + } + byte[] b = new byte[payloadSize]; + in.get(b); + message.setPayload(b); + + out.write(message); + return OK; + } + +} diff --git a/parser/src/main/java/org/dna/mqtt/moquette/proto/PublishEncoder.java b/parser/src/main/java/org/dna/mqtt/moquette/proto/PublishEncoder.java index 9ac028b8..b44dafa0 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/proto/PublishEncoder.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/proto/PublishEncoder.java @@ -1,48 +1,48 @@ -package org.dna.mqtt.moquette.proto; - -import org.apache.mina.core.buffer.IoBuffer; -import org.apache.mina.core.session.IoSession; -import org.apache.mina.filter.codec.ProtocolEncoderOutput; -import org.apache.mina.filter.codec.demux.MessageEncoder; -import org.dna.mqtt.moquette.proto.messages.AbstractMessage; -import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; -import org.dna.mqtt.moquette.proto.messages.PublishMessage; - -/** - * - * @author andrea - */ -public class PublishEncoder implements MessageEncoder { - - public void encode(IoSession session, PublishMessage message, ProtocolEncoderOutput out) throws Exception { - if (message.getQos() == QOSType.RESERVED) { - throw new IllegalArgumentException("Found a message with RESERVED Qos"); - } - if (message.getTopicName() == null || message.getTopicName().isEmpty()) { - throw new IllegalArgumentException("Found a message with empty or null topic name"); - } - - IoBuffer variableHeaderBuff = IoBuffer.allocate(2).setAutoExpand(true); - variableHeaderBuff.put(Utils.encodeString(message.getTopicName())); - if (message.getQos() == QOSType.LEAST_ONE || - message.getQos() == QOSType.EXACTLY_ONCE ) { - if (message.getMessageID() == null) { - throw new IllegalArgumentException("Found a message with QOS 1 or 2 and not MessageID setted"); - } - Utils.writeWord(variableHeaderBuff, message.getMessageID()); - } - variableHeaderBuff.put(message.getPayload()); - variableHeaderBuff.flip(); - int variableHeaderSize = variableHeaderBuff.remaining(); - - byte flags = Utils.encodeFlags(message); - - IoBuffer buff = IoBuffer.allocate(2 + variableHeaderSize); - buff.put((byte) (AbstractMessage.PUBLISH << 4 | flags)); - buff.put(Utils.encodeRemainingLength(variableHeaderSize)); - buff.put(variableHeaderBuff).flip(); - - out.write(buff); - } - -} +package org.dna.mqtt.moquette.proto; + +import org.apache.mina.core.buffer.IoBuffer; +import org.apache.mina.core.session.IoSession; +import org.apache.mina.filter.codec.ProtocolEncoderOutput; +import org.apache.mina.filter.codec.demux.MessageEncoder; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; +import org.dna.mqtt.moquette.proto.messages.PublishMessage; + +/** + * + * @author andrea + */ +public class PublishEncoder implements MessageEncoder { + + public void encode(IoSession session, PublishMessage message, ProtocolEncoderOutput out) throws Exception { + if (message.getQos() == QOSType.RESERVED) { + throw new IllegalArgumentException("Found a message with RESERVED Qos"); + } + if (message.getTopicName() == null || message.getTopicName().isEmpty()) { + throw new IllegalArgumentException("Found a message with empty or null topic name"); + } + + IoBuffer variableHeaderBuff = IoBuffer.allocate(2).setAutoExpand(true); + variableHeaderBuff.put(Utils.encodeString(message.getTopicName())); + if (message.getQos() == QOSType.LEAST_ONE || + message.getQos() == QOSType.EXACTLY_ONCE ) { + if (message.getMessageID() == null) { + throw new IllegalArgumentException("Found a message with QOS 1 or 2 and not MessageID setted"); + } + Utils.writeWord(variableHeaderBuff, message.getMessageID()); + } + variableHeaderBuff.put(message.getPayload()); + variableHeaderBuff.flip(); + int variableHeaderSize = variableHeaderBuff.remaining(); + + byte flags = Utils.encodeFlags(message); + + IoBuffer buff = IoBuffer.allocate(2 + variableHeaderSize).setAutoExpand(true);; + buff.put((byte) (AbstractMessage.PUBLISH << 4 | flags)); + buff.put(Utils.encodeRemainingLength(variableHeaderSize)); + buff.put(variableHeaderBuff).flip(); + + out.write(buff); + } + +} diff --git a/parser/src/main/java/org/dna/mqtt/moquette/proto/Utils.java b/parser/src/main/java/org/dna/mqtt/moquette/proto/Utils.java index b5dd63d8..899b5e7e 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/proto/Utils.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/proto/Utils.java @@ -1,180 +1,191 @@ -package org.dna.mqtt.moquette.proto; - -import java.io.UnsupportedEncodingException; -import org.apache.mina.core.buffer.IoBuffer; -import org.apache.mina.filter.codec.demux.MessageDecoderResult; -import org.dna.mqtt.moquette.proto.messages.AbstractMessage; -import org.slf4j.LoggerFactory; - -/** - * Common utils methodd used in codecs. - * - * @author andrea - */ -public class Utils { - - public static final int MAX_LENGTH_LIMIT = 268435455; - - /** - * Read 2 bytes from in buffer first MSB, and then LSB returning as int. - */ - static int readWord(IoBuffer in) { - int msb = in.get() & 0x00FF; //remove sign extension due to casting - int lsb = in.get() & 0x00FF; - msb = (msb << 8) | lsb ; - return msb; - } - - /** - * Writes as 2 bytes the int value into buffer first MSB, and then LSB. - */ - static void writeWord(IoBuffer out, int value) { - out.put((byte) ((value & 0xFF00) >> 8)); //msb - out.put((byte) (value & 0x00FF)); //lsb - } - - /** - * Decode the variable remaining lenght as defined in MQTT v3.1 specification - * (section 2.1). - * - * @return the decoded length or -1 if needed more data to decode the length field. - */ - static int decodeRemainingLenght(IoBuffer in) { - int multiplier = 1; - int value = 0; - byte digit; - do { - if (in.remaining() < 1) { - return -1; - } - digit = in.get(); - value += (digit & 0x7F) * multiplier; - multiplier *= 128; - } while ((digit & 0x80) != 0); - return value; - } - - /** - * Encode the value in the format defined in specification as variable length - * array. - * - * @throws IllegalArgumentException if the value is not in the specification bounds - * [0..268435455]. - */ - static IoBuffer encodeRemainingLength(int value) throws IllegalAccessException { - if (value > MAX_LENGTH_LIMIT || value < 0) { - throw new IllegalAccessException("Value should in range 0.." + MAX_LENGTH_LIMIT + " found " + value); - } - - IoBuffer encoded = IoBuffer.allocate(4); - byte digit; - do { - digit = (byte) (value % 128); - value = value / 128; - // if there are more digits to encode, set the top bit of this digit - if (value > 0) { - digit = (byte) (digit | 0x80); - } - encoded.put(digit); - } while (value > 0); - encoded.flip(); - return encoded; - } - - static MessageDecoderResult checkDecodable(byte type, IoBuffer in) { - if (in.remaining() < 1) { - return MessageDecoderResult.NEED_DATA; - } - byte h1 = in.get(); - byte messageType = (byte) ((h1 & 0x00F0) >> 4); - - int remainingLength = Utils.decodeRemainingLenght(in); - if (remainingLength == -1) { - return MessageDecoderResult.NEED_DATA; - } - - //check remaining length - if (in.remaining() < remainingLength) { - return MessageDecoderResult.NEED_DATA; - } - - return messageType == type ? MessageDecoderResult.OK : MessageDecoderResult.NOT_OK; - } - - /** - * Return the IoBuffer with string encoded as MSB, LSB and UTF-8 encoded - * string content. - */ - static IoBuffer encodeString(String str) { - IoBuffer out = IoBuffer.allocate(2).setAutoExpand(true); - byte[] raw; - try { - raw = str.getBytes("UTF-8"); - //NB every Java platform has got UTF-8 encoding by default, so this - //exception are never raised. - } catch (UnsupportedEncodingException ex) { - LoggerFactory.getLogger(ConnectEncoder.class).error(null, ex); - return null; - } - Utils.writeWord(out, raw.length); - out.put(raw).flip(); - return out; - } - - - /** - * Load a string from the given buffer, reading first the two bytes of len - * and then the UTF-8 bytes of the string. - * - * @return the decoded string or null if NEED_DATA - */ - static String decodeString(IoBuffer in) throws UnsupportedEncodingException { - if (in.remaining() < 2) { - return null; - } - int strLen = Utils.readWord(in); - if (in.remaining() < strLen) { - return null; - } - byte[] strRaw = new byte[strLen]; - in.get(strRaw); - - return new String(strRaw, "UTF-8"); - } - - static byte encodeFlags(AbstractMessage message) { - byte flags = 0; - if (message.isDupFlag()) { - flags |= 0x08; - } - if (message.isRetainFlag()) { - flags |= 0x01; - } - - flags |= ((message.getQos().ordinal() & 0x03) << 1); - return flags; - } - - /** - * Converts MQTT message type to a textual description. - * */ - public static String msgType2String(int type) { - switch (type) { - case AbstractMessage.CONNECT: return "CONNECT"; - case AbstractMessage.CONNACK: return "CONNACK"; - case AbstractMessage.PUBLISH: return "PUBLISH"; - case AbstractMessage.PUBACK: return "PUBACK"; - case AbstractMessage.PUBREC: return "PUBREC"; - case AbstractMessage.PUBREL: return "PUBREL"; - case AbstractMessage.PUBCOMP: return "PUBCOMP"; - case AbstractMessage.SUBSCRIBE: return "SUBSCRIBE"; - case AbstractMessage.SUBACK: return "SUBACK"; - case AbstractMessage.UNSUBSCRIBE: return "UNSUBSCRIBE"; - case AbstractMessage.UNSUBACK: return "UNSUBACK"; - case AbstractMessage.PINGREQ: return "PINGREQ"; - case AbstractMessage.PINGRESP: return "PINGRESP"; - case AbstractMessage.DISCONNECT: return "DISCONNECT"; - default: throw new RuntimeException("Can't decode message type " + type); - } - } -} +package org.dna.mqtt.moquette.proto; + +import java.io.UnsupportedEncodingException; +import org.apache.mina.core.buffer.IoBuffer; +import org.apache.mina.filter.codec.demux.MessageDecoderResult; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.slf4j.LoggerFactory; + +/** + * Common utils methodd used in codecs. + * + * @author andrea + */ +public class Utils { + + public static final int MAX_LENGTH_LIMIT = 268435455; + + /** + * Read 2 bytes from in buffer first MSB, and then LSB returning as int. + */ + static int readWord(IoBuffer in) { + int msb = in.get() & 0x00FF; //remove sign extension due to casting + int lsb = in.get() & 0x00FF; + msb = (msb << 8) | lsb ; + return msb; + } + + /** + * Writes as 2 bytes the int value into buffer first MSB, and then LSB. + */ + static void writeWord(IoBuffer out, int value) { + out.put((byte) ((value & 0xFF00) >> 8)); //msb + out.put((byte) (value & 0x00FF)); //lsb + } + + /** + * Decode the variable remaining lenght as defined in MQTT v3.1 specification + * (section 2.1). + * + * @return the decoded length or -1 if needed more data to decode the length field. + */ + static int decodeRemainingLenght(IoBuffer in) { + int multiplier = 1; + int value = 0; + byte digit; + do { + if (in.remaining() < 1) { + return -1; + } + digit = in.get(); + value += (digit & 0x7F) * multiplier; + multiplier *= 128; + } while ((digit & 0x80) != 0); + return value; + } + + /** + * Return the number of bytes to encode the gicen remaining length value + */ + static int numBytesToEncode(int len) { + if (0 <= len && len <= 127) return 1; + if (128 <= len && len <= 16383) return 2; + if (16384 <= len && len <= 2097151) return 3; + if (2097152 <= len && len <= 268435455) return 4; + throw new IllegalArgumentException("value shoul be in the range [0..268435455]"); + } + + /** + * Encode the value in the format defined in specification as variable length + * array. + * + * @throws IllegalArgumentException if the value is not in the specification bounds + * [0..268435455]. + */ + static IoBuffer encodeRemainingLength(int value) throws IllegalAccessException { + if (value > MAX_LENGTH_LIMIT || value < 0) { + throw new IllegalAccessException("Value should in range 0.." + MAX_LENGTH_LIMIT + " found " + value); + } + + IoBuffer encoded = IoBuffer.allocate(4); + byte digit; + do { + digit = (byte) (value % 128); + value = value / 128; + // if there are more digits to encode, set the top bit of this digit + if (value > 0) { + digit = (byte) (digit | 0x80); + } + encoded.put(digit); + } while (value > 0); + encoded.flip(); + return encoded; + } + + static MessageDecoderResult checkDecodable(byte type, IoBuffer in) { + if (in.remaining() < 1) { + return MessageDecoderResult.NEED_DATA; + } + byte h1 = in.get(); + byte messageType = (byte) ((h1 & 0x00F0) >> 4); + + int remainingLength = Utils.decodeRemainingLenght(in); + if (remainingLength == -1) { + return MessageDecoderResult.NEED_DATA; + } + + //check remaining length + if (in.remaining() < remainingLength) { + return MessageDecoderResult.NEED_DATA; + } + + return messageType == type ? MessageDecoderResult.OK : MessageDecoderResult.NOT_OK; + } + + /** + * Return the IoBuffer with string encoded as MSB, LSB and UTF-8 encoded + * string content. + */ + static IoBuffer encodeString(String str) { + IoBuffer out = IoBuffer.allocate(2).setAutoExpand(true); + byte[] raw; + try { + raw = str.getBytes("UTF-8"); + //NB every Java platform has got UTF-8 encoding by default, so this + //exception are never raised. + } catch (UnsupportedEncodingException ex) { + LoggerFactory.getLogger(ConnectEncoder.class).error(null, ex); + return null; + } + Utils.writeWord(out, raw.length); + out.put(raw).flip(); + return out; + } + + + /** + * Load a string from the given buffer, reading first the two bytes of len + * and then the UTF-8 bytes of the string. + * + * @return the decoded string or null if NEED_DATA + */ + static String decodeString(IoBuffer in) throws UnsupportedEncodingException { + if (in.remaining() < 2) { + return null; + } + int strLen = Utils.readWord(in); + if (in.remaining() < strLen) { + return null; + } + byte[] strRaw = new byte[strLen]; + in.get(strRaw); + + return new String(strRaw, "UTF-8"); + } + + static byte encodeFlags(AbstractMessage message) { + byte flags = 0; + if (message.isDupFlag()) { + flags |= 0x08; + } + if (message.isRetainFlag()) { + flags |= 0x01; + } + + flags |= ((message.getQos().ordinal() & 0x03) << 1); + return flags; + } + + /** + * Converts MQTT message type to a textual description. + * */ + public static String msgType2String(int type) { + switch (type) { + case AbstractMessage.CONNECT: return "CONNECT"; + case AbstractMessage.CONNACK: return "CONNACK"; + case AbstractMessage.PUBLISH: return "PUBLISH"; + case AbstractMessage.PUBACK: return "PUBACK"; + case AbstractMessage.PUBREC: return "PUBREC"; + case AbstractMessage.PUBREL: return "PUBREL"; + case AbstractMessage.PUBCOMP: return "PUBCOMP"; + case AbstractMessage.SUBSCRIBE: return "SUBSCRIBE"; + case AbstractMessage.SUBACK: return "SUBACK"; + case AbstractMessage.UNSUBSCRIBE: return "UNSUBSCRIBE"; + case AbstractMessage.UNSUBACK: return "UNSUBACK"; + case AbstractMessage.PINGREQ: return "PINGREQ"; + case AbstractMessage.PINGRESP: return "PINGRESP"; + case AbstractMessage.DISCONNECT: return "DISCONNECT"; + default: throw new RuntimeException("Can't decode message type " + type); + } + } +} diff --git a/parser/src/test/java/org/dna/mqtt/moquette/proto/PublishDecoderTest.java b/parser/src/test/java/org/dna/mqtt/moquette/proto/PublishDecoderTest.java index 7fcf0be5..d4e0510f 100644 --- a/parser/src/test/java/org/dna/mqtt/moquette/proto/PublishDecoderTest.java +++ b/parser/src/test/java/org/dna/mqtt/moquette/proto/PublishDecoderTest.java @@ -1,132 +1,180 @@ -package org.dna.mqtt.moquette.proto; - -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.messages.AbstractMessage; -import org.dna.mqtt.moquette.proto.messages.PublishMessage; -import static org.junit.Assert.*; -import org.junit.Before; -import org.junit.Test; - -/** - * - * @author andrea - */ -public class PublishDecoderTest { - - IoBuffer m_buff; - PublishDecoder m_msgdec; - MockProtocolDecoderOutput m_mockProtoDecoder; - - @Before - public void setUp() { - m_buff = IoBuffer.allocate(4); - m_msgdec = new PublishDecoder(); - m_mockProtoDecoder = new MockProtocolDecoderOutput(); - } - - @Test - public void testDecodable_OK() { - m_buff.put((byte) (AbstractMessage.PUBLISH << 4)) - .put((byte)0) //0 length - .flip(); - assertEquals(MessageDecoder.OK, m_msgdec.decodable(null, m_buff)); - } - - @Test - public void testHeader() throws Exception { - m_buff = IoBuffer.allocate(14); - initHeader(m_buff); - m_buff.flip(); - - //Exercise - MessageDecoderResult res = m_msgdec.decode(null, m_buff, m_mockProtoDecoder); - - assertNotNull(m_mockProtoDecoder.getMessage()); - assertEquals(MessageDecoder.OK, res); - assertEquals("Fake Topic", m_mockProtoDecoder.getMessage().getTopicName()); - assertNull(m_mockProtoDecoder.getMessage().getMessageID()); - assertEquals(AbstractMessage.PUBLISH, m_mockProtoDecoder.getMessage().getMessageType()); - } - - @Test - public void testHeaderWithMessageID() throws Exception { - m_buff = IoBuffer.allocate(14).setAutoExpand(true); - int messageID = 123; - initHeaderWithMessageID(m_buff, messageID); - m_buff.flip(); - - //Exercise - MessageDecoderResult res = m_msgdec.decode(null, m_buff, m_mockProtoDecoder); - - assertNotNull(m_mockProtoDecoder.getMessage()); - assertEquals(MessageDecoder.OK, res); - assertEquals("Fake Topic", m_mockProtoDecoder.getMessage().getTopicName()); - assertEquals(messageID, (int) m_mockProtoDecoder.getMessage().getMessageID()); - } - - @Test - public void testHeaderWithMessageID_Payload() throws Exception { - m_buff = IoBuffer.allocate(14).setAutoExpand(true); - int messageID = 123; - byte[] payload = new byte[]{0x0A, 0x0B, 0x0C}; - initHeaderWithMessageID_Payload(m_buff, messageID, payload); - m_buff.flip(); - - //Exercise - MessageDecoderResult res = m_msgdec.decode(null, m_buff, m_mockProtoDecoder); - - assertNotNull(m_mockProtoDecoder.getMessage()); - assertEquals(MessageDecoder.OK, res); - assertEquals("Fake Topic", m_mockProtoDecoder.getMessage().getTopicName()); - assertEquals(messageID, (int) m_mockProtoDecoder.getMessage().getMessageID()); - TestUtils.verifyEquals(payload, m_mockProtoDecoder.getMessage().getPayload()); - } - - @Test - public void testBugOnRealCase() throws Exception { - byte[] overallMessage = new byte[] {0x30, 0x17, //fixed header, 25 byte lenght - 0x00, 0x06, 0x2f, 0x74, 0x6f, 0x70, 0x69, 0x63, //[/topic] 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 = IoBuffer.allocate(overallMessage.length).setAutoExpand(true); - m_buff.put(overallMessage); - m_buff.flip(); - - //Exercise - MessageDecoderResult res = m_msgdec.decode(null, m_buff, m_mockProtoDecoder); - - assertNotNull(m_mockProtoDecoder.getMessage()); - assertEquals(MessageDecoder.OK, res); - } - - private void initHeader(IoBuffer buff) throws IllegalAccessException { - IoBuffer tmp = IoBuffer.allocate(4).setAutoExpand(true).put(Utils.encodeString("Fake Topic")).flip(); - buff.clear().put((byte) (AbstractMessage.PUBLISH << 4)).put(Utils.encodeRemainingLength(tmp.remaining())); - //topic name - buff.put(tmp); - } - - private void initHeaderWithMessageID(IoBuffer buff, int messageID) throws IllegalAccessException { - IoBuffer tmp = IoBuffer.allocate(4).setAutoExpand(true).put(Utils.encodeString("Fake Topic")); - Utils.writeWord(tmp, messageID); - tmp.flip(); - buff.clear().put((byte) (AbstractMessage.PUBLISH << 4 | 0x02)) //set Qos to 1 - .put(Utils.encodeRemainingLength(tmp.remaining())); - //topic name - buff.put(tmp); - } - - private void initHeaderWithMessageID_Payload(IoBuffer buff, int messageID, byte[] payload) throws IllegalAccessException { - IoBuffer tmp = IoBuffer.allocate(4).setAutoExpand(true).put(Utils.encodeString("Fake Topic")); - Utils.writeWord(tmp, messageID); - tmp.put(payload); - tmp.flip(); - buff.clear().put((byte) (AbstractMessage.PUBLISH << 4 | 0x02)) //set Qos to 1 - .put(Utils.encodeRemainingLength(tmp.remaining())); - //topic name - buff.put(tmp); - } -} +package org.dna.mqtt.moquette.proto; + +import java.io.File; +import java.io.FileInputStream; +import java.io.FileNotFoundException; +import java.io.IOException; +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.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.PublishMessage; +import static org.junit.Assert.*; +import org.junit.Before; +import org.junit.Ignore; +import org.junit.Test; + +/** + * + * @author andrea + */ +public class PublishDecoderTest { + + IoBuffer m_buff; + PublishDecoder m_msgdec; + MockProtocolDecoderOutput m_mockProtoDecoder; + + @Before + public void setUp() { + m_buff = IoBuffer.allocate(4); + m_msgdec = new PublishDecoder(); + m_mockProtoDecoder = new MockProtocolDecoderOutput(); + } + + @Test + public void testDecodable_OK() { + m_buff.put((byte) (AbstractMessage.PUBLISH << 4)) + .put((byte) 0) //0 length + .flip(); + assertEquals(MessageDecoder.OK, m_msgdec.decodable(null, m_buff)); + } + + @Test + public void testHeader() throws Exception { + m_buff = IoBuffer.allocate(14); + initHeader(m_buff); + m_buff.flip(); + + //Exercise + MessageDecoderResult res = m_msgdec.decode(null, m_buff, m_mockProtoDecoder); + + assertNotNull(m_mockProtoDecoder.getMessage()); + assertEquals(MessageDecoder.OK, res); + assertEquals("Fake Topic", m_mockProtoDecoder.getMessage().getTopicName()); + assertNull(m_mockProtoDecoder.getMessage().getMessageID()); + assertEquals(AbstractMessage.PUBLISH, m_mockProtoDecoder.getMessage().getMessageType()); + } + + @Test + public void testHeaderWithMessageID() throws Exception { + m_buff = IoBuffer.allocate(14).setAutoExpand(true); + int messageID = 123; + initHeaderWithMessageID(m_buff, messageID); + m_buff.flip(); + + //Exercise + MessageDecoderResult res = m_msgdec.decode(null, m_buff, m_mockProtoDecoder); + + assertNotNull(m_mockProtoDecoder.getMessage()); + assertEquals(MessageDecoder.OK, res); + assertEquals("Fake Topic", m_mockProtoDecoder.getMessage().getTopicName()); + assertEquals(messageID, (int) m_mockProtoDecoder.getMessage().getMessageID()); + } + + @Test + public void testHeaderWithMessageID_Payload() throws Exception { + m_buff = IoBuffer.allocate(14).setAutoExpand(true); + int messageID = 123; + byte[] payload = new byte[]{0x0A, 0x0B, 0x0C}; + initHeaderWithMessageID_Payload(m_buff, messageID, payload); + m_buff.flip(); + + //Exercise + MessageDecoderResult res = m_msgdec.decode(null, m_buff, m_mockProtoDecoder); + + assertNotNull(m_mockProtoDecoder.getMessage()); + assertEquals(MessageDecoder.OK, res); + assertEquals("Fake Topic", m_mockProtoDecoder.getMessage().getTopicName()); + assertEquals(messageID, (int) m_mockProtoDecoder.getMessage().getMessageID()); + TestUtils.verifyEquals(payload, m_mockProtoDecoder.getMessage().getPayload()); + } + + @Test + public void testBugOnRealCase() throws Exception { + byte[] overallMessage = new byte[]{0x30, 0x17, //fixed header, 25 byte lenght + 0x00, 0x06, 0x2f, 0x74, 0x6f, 0x70, 0x69, 0x63, //[/topic] 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 = IoBuffer.allocate(overallMessage.length).setAutoExpand(true); + m_buff.put(overallMessage); + m_buff.flip(); + + //Exercise + MessageDecoderResult res = m_msgdec.decode(null, m_buff, m_mockProtoDecoder); + + assertNotNull(m_mockProtoDecoder.getMessage()); + assertEquals(MessageDecoder.OK, res); + } + + @Test + public void testDecodeBigContent() throws Exception { + int size = 129; + IoBuffer payload = TestUtils.generateRandomPayload(size); + + IoBuffer firstPublish = generatePublishQoS0(payload); + IoBuffer secondPublish = generatePublishQoS0(TestUtils.generateRandomPayload(size)); + + IoBuffer doubleMessageBuf = IoBuffer.allocate(size * 2).setAutoExpand(true); + doubleMessageBuf.put(firstPublish).put(secondPublish).flip(); + + + //Exercise + MessageDecoderResult res = m_msgdec.decode(null, doubleMessageBuf, m_mockProtoDecoder); + + assertEquals(MessageDecoder.OK, res); + PublishMessage pubMsg = m_mockProtoDecoder.getMessage(); + assertNotNull(pubMsg); + + res = m_msgdec.decode(null, doubleMessageBuf, m_mockProtoDecoder); + + assertNotNull(m_mockProtoDecoder.getMessage()); + assertEquals(MessageDecoder.OK, res); + + m_buff.flip(); + } + + private void initHeader(IoBuffer buff) throws IllegalAccessException { + IoBuffer tmp = IoBuffer.allocate(4).setAutoExpand(true).put(Utils.encodeString("Fake Topic")).flip(); + buff.clear().put((byte) (AbstractMessage.PUBLISH << 4)).put(Utils.encodeRemainingLength(tmp.remaining())); + //topic name + buff.put(tmp); + } + + private void initHeaderWithMessageID(IoBuffer buff, int messageID) throws IllegalAccessException { + IoBuffer tmp = IoBuffer.allocate(4).setAutoExpand(true).put(Utils.encodeString("Fake Topic")); + Utils.writeWord(tmp, messageID); + tmp.flip(); + buff.clear().put((byte) (AbstractMessage.PUBLISH << 4 | 0x02)) //set Qos to 1 + .put(Utils.encodeRemainingLength(tmp.remaining())); + //topic name + buff.put(tmp); + } + + private void initHeaderWithMessageID_Payload(IoBuffer buff, int messageID, byte[] payload) throws IllegalAccessException { + IoBuffer tmp = IoBuffer.allocate(4).setAutoExpand(true).put(Utils.encodeString("Fake Topic")); + Utils.writeWord(tmp, messageID); + tmp.put(payload); + tmp.flip(); + buff.clear().put((byte) (AbstractMessage.PUBLISH << 4 | 0x02)) //set Qos to 1 + .put(Utils.encodeRemainingLength(tmp.remaining())); + //topic name + buff.put(tmp); + } + + private IoBuffer generatePublishQoS0(IoBuffer payload) throws IllegalAccessException { + int size = payload.capacity(); + IoBuffer messageBody = IoBuffer.allocate(size).setAutoExpand(true); + messageBody.put(Utils.encodeString("/topic")); + + //ONLY for QoS > 1 Utils.writeWord(messageBody, messageID); + messageBody.put(payload).flip(); + + IoBuffer completeMsg = IoBuffer.allocate(size).setAutoExpand(true); + completeMsg.clear().put((byte) (AbstractMessage.PUBLISH << 4 | 0x00)) //set Qos to 0 + .put(Utils.encodeRemainingLength(messageBody.remaining())); + completeMsg.put(messageBody).flip(); + + return completeMsg; + } +} diff --git a/parser/src/test/java/org/dna/mqtt/moquette/proto/PublishEncoderTest.java b/parser/src/test/java/org/dna/mqtt/moquette/proto/PublishEncoderTest.java index 6aeac635..c3dd9f15 100644 --- a/parser/src/test/java/org/dna/mqtt/moquette/proto/PublishEncoderTest.java +++ b/parser/src/test/java/org/dna/mqtt/moquette/proto/PublishEncoderTest.java @@ -1,145 +1,167 @@ -package org.dna.mqtt.moquette.proto; - -import org.dna.mqtt.moquette.proto.PublishEncoder; -import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; -import org.dna.mqtt.moquette.proto.messages.PublishMessage; -import org.junit.Before; -import org.junit.Test; -import static org.junit.Assert.*; -import static org.dna.mqtt.moquette.proto.TestUtils.*; - -/** - * - * @author andrea - */ -public class PublishEncoderTest { - - PublishEncoder m_encoder; - TestUtils.MockProtocolEncoderOutput m_mockProtoEncoder; - - @Before - public void setUp() { - m_encoder = new PublishEncoder(); - m_mockProtoEncoder = new TestUtils.MockProtocolEncoderOutput(); - } - - @Test - public void testEncodeWithQos_0_noMessageID() throws Exception { - String topic = "/photos"; - PublishMessage msg = new PublishMessage(); - msg.setQos(QOSType.MOST_ONE); - msg.setTopicName(topic); - - //variable part - byte[] payload = new byte[]{0x0A, 0x0B, 0x0C}; - msg.setPayload(payload); - - //Exercise - m_encoder.encode(null, msg, m_mockProtoEncoder); - - //Verify - assertEquals(0x30, m_mockProtoEncoder.getBuffer().get()); //1 byte - //(2+7) topic + 3 payload - assertEquals(12, m_mockProtoEncoder.getBuffer().get()); //remaining length - - //Variable part - verifyString(topic, m_mockProtoEncoder.getBuffer()); - verifyBuff(payload.length, payload, m_mockProtoEncoder.getBuffer()); - } - - @Test - public void testEncodeWithQos_1_MessageID() throws Exception { - String topic = "/photos"; - PublishMessage msg = new PublishMessage(); - msg.setQos(QOSType.LEAST_ONE); - msg.setMessageID(1); - msg.setTopicName(topic); - - //variable part - byte[] payload = new byte[]{0x0A, 0x0B, 0x0C}; - msg.setPayload(payload); - - //Exercise - m_encoder.encode(null, msg, m_mockProtoEncoder); - - //Verify - assertEquals(0x32, m_mockProtoEncoder.getBuffer().get()); //1 byte - //(2+7) topic + 2 messageID + 3 payload - assertEquals(14, m_mockProtoEncoder.getBuffer().get()); //remaining length - - //Variable part - verifyString(topic, m_mockProtoEncoder.getBuffer()); - assertEquals(0, m_mockProtoEncoder.getBuffer().get()); //MessageID MSB - assertEquals(1, m_mockProtoEncoder.getBuffer().get()); //MessageID LSB - verifyBuff(payload.length, payload, m_mockProtoEncoder.getBuffer()); - } - - @Test(expected = IllegalArgumentException.class) - public void testEncodeWithQos_1_noMessageID_fail() throws Exception { - String topic = "pictures/photos"; - PublishMessage msg = new PublishMessage(); - msg.setQos(QOSType.LEAST_ONE); - msg.setTopicName(topic); - - //Exercise - m_encoder.encode(null, msg, m_mockProtoEncoder); - } - - @Test(expected = IllegalArgumentException.class) - public void testEncode_empty_topic() throws Exception { - PublishMessage msg = new PublishMessage(); - msg.setQos(QOSType.LEAST_ONE); - msg.setMessageID(1); - - //Exercise - m_encoder.encode(null, msg, m_mockProtoEncoder); - } - - @Test(expected = IllegalArgumentException.class) - public void testEncode_null_topic() throws Exception { - PublishMessage msg = new PublishMessage(); - msg.setQos(QOSType.LEAST_ONE); - msg.setMessageID(1); - msg.setTopicName(null); - - //Exercise - m_encoder.encode(null, msg, m_mockProtoEncoder); - } - - @Test(expected = IllegalArgumentException.class) - public void testEncode_qos_reserved() throws Exception { - PublishMessage msg = new PublishMessage(); - msg.setQos(QOSType.RESERVED); - msg.setMessageID(1); - msg.setTopicName(null); - - //variable part - byte[] payload = new byte[]{0x0A, 0x0B, 0x0C}; - msg.setPayload(payload); - - //Exercise - m_encoder.encode(null, msg, m_mockProtoEncoder); - } - - @Test - public void testBugMissedMessageIDEncoding() throws Exception{ - String topic = "/topic"; - PublishMessage msg = new PublishMessage(); - msg.setQos(QOSType.MOST_ONE); - msg.setTopicName(topic); - - //variable part - byte[] payload = "Test my payload".getBytes(); - msg.setPayload(payload); - - //Exercise - m_encoder.encode(null, msg, m_mockProtoEncoder); - - //Verify - //2 byte header + (2+6) topic name + [2 message ID if QoQ <> 0]+ payload - int size = m_mockProtoEncoder.getBuffer().remaining(); - assertEquals(10+payload.length, size); - assertEquals(0x30, m_mockProtoEncoder.getBuffer().get()); //1 byte - assertEquals(23, m_mockProtoEncoder.getBuffer().get()); //1 byte the length - } -} +package org.dna.mqtt.moquette.proto; + +import org.apache.mina.core.buffer.IoBuffer; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; +import org.dna.mqtt.moquette.proto.messages.PublishMessage; +import org.junit.Before; +import org.junit.Test; +import static org.junit.Assert.*; +import static org.dna.mqtt.moquette.proto.TestUtils.*; + +/** + * + * @author andrea + */ +public class PublishEncoderTest { + + PublishEncoder m_encoder; + TestUtils.MockProtocolEncoderOutput m_mockProtoEncoder; + + @Before + public void setUp() { + m_encoder = new PublishEncoder(); + m_mockProtoEncoder = new TestUtils.MockProtocolEncoderOutput(); + } + + @Test + public void testEncodeWithQos_0_noMessageID() throws Exception { + String topic = "/photos"; + PublishMessage msg = new PublishMessage(); + msg.setQos(QOSType.MOST_ONE); + msg.setTopicName(topic); + + //variable part + byte[] payload = new byte[]{0x0A, 0x0B, 0x0C}; + msg.setPayload(payload); + + //Exercise + m_encoder.encode(null, msg, m_mockProtoEncoder); + + //Verify + assertEquals(0x30, m_mockProtoEncoder.getBuffer().get()); //1 byte + //(2+7) topic + 3 payload + assertEquals(12, m_mockProtoEncoder.getBuffer().get()); //remaining length + + //Variable part + verifyString(topic, m_mockProtoEncoder.getBuffer()); + verifyBuff(payload.length, payload, m_mockProtoEncoder.getBuffer()); + } + + @Test + public void testEncodeWithQos_1_MessageID() throws Exception { + String topic = "/photos"; + PublishMessage msg = new PublishMessage(); + msg.setQos(QOSType.LEAST_ONE); + msg.setMessageID(1); + msg.setTopicName(topic); + + //variable part + byte[] payload = new byte[]{0x0A, 0x0B, 0x0C}; + msg.setPayload(payload); + + //Exercise + m_encoder.encode(null, msg, m_mockProtoEncoder); + + //Verify + assertEquals(0x32, m_mockProtoEncoder.getBuffer().get()); //1 byte + //(2+7) topic + 2 messageID + 3 payload + assertEquals(14, m_mockProtoEncoder.getBuffer().get()); //remaining length + + //Variable part + verifyString(topic, m_mockProtoEncoder.getBuffer()); + assertEquals(0, m_mockProtoEncoder.getBuffer().get()); //MessageID MSB + assertEquals(1, m_mockProtoEncoder.getBuffer().get()); //MessageID LSB + verifyBuff(payload.length, payload, m_mockProtoEncoder.getBuffer()); + } + + @Test(expected = IllegalArgumentException.class) + public void testEncodeWithQos_1_noMessageID_fail() throws Exception { + String topic = "pictures/photos"; + PublishMessage msg = new PublishMessage(); + msg.setQos(QOSType.LEAST_ONE); + msg.setTopicName(topic); + + //Exercise + m_encoder.encode(null, msg, m_mockProtoEncoder); + } + + @Test(expected = IllegalArgumentException.class) + public void testEncode_empty_topic() throws Exception { + PublishMessage msg = new PublishMessage(); + msg.setQos(QOSType.LEAST_ONE); + msg.setMessageID(1); + + //Exercise + m_encoder.encode(null, msg, m_mockProtoEncoder); + } + + @Test(expected = IllegalArgumentException.class) + public void testEncode_null_topic() throws Exception { + PublishMessage msg = new PublishMessage(); + msg.setQos(QOSType.LEAST_ONE); + msg.setMessageID(1); + msg.setTopicName(null); + + //Exercise + m_encoder.encode(null, msg, m_mockProtoEncoder); + } + + @Test(expected = IllegalArgumentException.class) + public void testEncode_qos_reserved() throws Exception { + PublishMessage msg = new PublishMessage(); + msg.setQos(QOSType.RESERVED); + msg.setMessageID(1); + msg.setTopicName(null); + + //variable part + byte[] payload = new byte[]{0x0A, 0x0B, 0x0C}; + msg.setPayload(payload); + + //Exercise + m_encoder.encode(null, msg, m_mockProtoEncoder); + } + + @Test + public void testBugMissedMessageIDEncoding() throws Exception{ + String topic = "/topic"; + PublishMessage msg = new PublishMessage(); + msg.setQos(QOSType.MOST_ONE); + msg.setTopicName(topic); + + //variable part + byte[] payload = "Test my payload".getBytes(); + msg.setPayload(payload); + + //Exercise + m_encoder.encode(null, msg, m_mockProtoEncoder); + + //Verify + //2 byte header + (2+6) topic name + [2 message ID if QoQ <> 0]+ payload + int size = m_mockProtoEncoder.getBuffer().remaining(); + assertEquals(10+payload.length, size); + assertEquals(0x30, m_mockProtoEncoder.getBuffer().get()); //1 byte + assertEquals(23, m_mockProtoEncoder.getBuffer().get()); //1 byte the length + } + + @Test + public void testEncodeBigMessage() throws Exception { + byte[] payload = new byte[256]; + + TestUtils.generateRandomPayload(256).get(payload); + + String topic = "/topic"; + PublishMessage msg = new PublishMessage(); + msg.setQos(QOSType.MOST_ONE); + msg.setTopicName(topic); + + //variable part + msg.setPayload(payload); + + //Exercise + m_encoder.encode(null, msg, m_mockProtoEncoder); + + int size = m_mockProtoEncoder.getBuffer().remaining(); + assertEquals(11+payload.length, size); + assertEquals(0x30, m_mockProtoEncoder.getBuffer().get()); //1 byte + } +} diff --git a/parser/src/test/java/org/dna/mqtt/moquette/proto/TestUtils.java b/parser/src/test/java/org/dna/mqtt/moquette/proto/TestUtils.java index 71d0d17e..fb4bb602 100644 --- a/parser/src/test/java/org/dna/mqtt/moquette/proto/TestUtils.java +++ b/parser/src/test/java/org/dna/mqtt/moquette/proto/TestUtils.java @@ -1,97 +1,124 @@ -package org.dna.mqtt.moquette.proto; - -import java.io.UnsupportedEncodingException; -import org.apache.mina.core.buffer.IoBuffer; -import org.apache.mina.core.filterchain.IoFilter.NextFilter; -import org.apache.mina.core.future.WriteFuture; -import org.apache.mina.core.session.IoSession; -import org.apache.mina.filter.codec.ProtocolDecoderOutput; -import org.apache.mina.filter.codec.ProtocolEncoderOutput; -import org.dna.mqtt.moquette.proto.messages.AbstractMessage; -import static org.junit.Assert.*; - -/** - * - * @author andrea - */ -public class TestUtils { - - static final class MockProtocolDecoderOutput implements ProtocolDecoderOutput { - - protected T m_connMessage; - - public T getMessage() { - return m_connMessage; - } - - public void write(Object message) { - m_connMessage = (T) message; - } - - public void flush(NextFilter nextFilter, IoSession session) { - throw new UnsupportedOperationException("Not supported yet."); - } - } - - - static final class MockProtocolEncoderOutput implements ProtocolEncoderOutput { - - protected IoBuffer m_buffer; - - public IoBuffer getBuffer() { - return m_buffer; - } - - public void write(Object buffer) { - m_buffer = (IoBuffer) buffer; - } - - public void mergeAll() { - throw new UnsupportedOperationException("Not supported yet."); - } - - public WriteFuture flush() { - throw new UnsupportedOperationException("Not supported yet."); - } - } - - /** - * Verify the presence of the given string starting from the current position - * inside the buffer. - */ - static void verifyString(String str, IoBuffer buff) throws UnsupportedEncodingException { - IoBuffer tmpBuff = IoBuffer.allocate(2).setAutoExpand(true); - byte[] raw = str.getBytes("UTF-8"); - Utils.writeWord(tmpBuff, raw.length); - tmpBuff.put(raw).flip(); - - verifyIoBuffer(tmpBuff, buff); - } - - static void verifyIoBuffer(IoBuffer expected, IoBuffer found) { - while(expected.hasRemaining()) { - assertEquals(expected.get(), found.get()); - } - } - - /** - * Verify that the given bytes buffer og the given numBytes length is present - * in the buff starting from the current position. - */ - static void verifyBuff(int numBytes, byte[] bytes, IoBuffer buff) { - assertTrue(numBytes <= buff.remaining()); - byte[] toCheck = new byte[numBytes]; - buff.get(toCheck); - - for (int i = 0; i < numBytes; i++) { - assertEquals(bytes[i], toCheck[i]); - } - } - - static void verifyEquals(byte[] expected, byte[] found) { - assertEquals(expected.length, found.length); - for (int i = 0; i < expected.length; i++) { - assertEquals(expected[i], found[i]); - } - } -} +package org.dna.mqtt.moquette.proto; + +import java.io.File; +import java.io.FileInputStream; +import java.io.FileNotFoundException; +import java.io.IOException; +import java.io.UnsupportedEncodingException; +import org.apache.mina.core.buffer.IoBuffer; +import org.apache.mina.core.filterchain.IoFilter.NextFilter; +import org.apache.mina.core.future.WriteFuture; +import org.apache.mina.core.session.IoSession; +import org.apache.mina.filter.codec.ProtocolDecoderOutput; +import org.apache.mina.filter.codec.ProtocolEncoderOutput; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import static org.junit.Assert.*; + +/** + * + * @author andrea + */ +public class TestUtils { + + static final class MockProtocolDecoderOutput implements ProtocolDecoderOutput { + + protected T m_connMessage; + + public T getMessage() { + return m_connMessage; + } + + public void write(Object message) { + m_connMessage = (T) message; + } + + public void flush(NextFilter nextFilter, IoSession session) { + throw new UnsupportedOperationException("Not supported yet."); + } + } + + + static final class MockProtocolEncoderOutput implements ProtocolEncoderOutput { + + protected IoBuffer m_buffer; + + public IoBuffer getBuffer() { + return m_buffer; + } + + public void write(Object buffer) { + m_buffer = (IoBuffer) buffer; + } + + public void mergeAll() { + throw new UnsupportedOperationException("Not supported yet."); + } + + public WriteFuture flush() { + throw new UnsupportedOperationException("Not supported yet."); + } + } + + /** + * Verify the presence of the given string starting from the current position + * inside the buffer. + */ + static void verifyString(String str, IoBuffer buff) throws UnsupportedEncodingException { + IoBuffer tmpBuff = IoBuffer.allocate(2).setAutoExpand(true); + byte[] raw = str.getBytes("UTF-8"); + Utils.writeWord(tmpBuff, raw.length); + tmpBuff.put(raw).flip(); + + verifyIoBuffer(tmpBuff, buff); + } + + static void verifyIoBuffer(IoBuffer expected, IoBuffer found) { + while(expected.hasRemaining()) { + assertEquals(expected.get(), found.get()); + } + } + + /** + * Verify that the given bytes buffer og the given numBytes length is present + * in the buff starting from the current position. + */ + static void verifyBuff(int numBytes, byte[] bytes, IoBuffer buff) { + assertTrue(numBytes <= buff.remaining()); + byte[] toCheck = new byte[numBytes]; + buff.get(toCheck); + + for (int i = 0; i < numBytes; i++) { + assertEquals(bytes[i], toCheck[i]); + } + } + + static void verifyEquals(byte[] expected, byte[] found) { + assertEquals(expected.length, found.length); + for (int i = 0; i < expected.length; i++) { + assertEquals(expected[i], found[i]); + } + } + + + public static IoBuffer generateRandomPayload(int size) { + IoBuffer payloadBuffer = IoBuffer.allocate(size); + for (int i = 0; i < size; i++) { + payloadBuffer.put((byte) (Math.random() * 255)); + } + payloadBuffer.flip(); + return payloadBuffer; + } + + public static IoBuffer loadFile(String path) throws FileNotFoundException, IOException { + File f = new File(path); + long size = f.length(); + byte[] payload = new byte[(int) size]; + + FileInputStream fr = new FileInputStream(f); + fr.read(payload); + + IoBuffer payloadBuffer = IoBuffer.allocate((int) size); + payloadBuffer.put(payload).flip(); + return payloadBuffer; + } +}