diff --git a/netty_parser/pom.xml b/netty_parser/pom.xml
index 3e1bee64..27276ef7 100644
--- a/netty_parser/pom.xml
+++ b/netty_parser/pom.xml
@@ -23,7 +23,7 @@
org.dna.mqtt
moquette-parser-commons
- 0.2
+ ${project.version}
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 00e869f8..593d8a5b 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
@@ -1,6 +1,7 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import java.util.List;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
@@ -51,9 +52,10 @@ class PublishDecoder extends DemuxDecoder {
in.resetReaderIndex();
return;
}
- byte[] b = new byte[payloadSize];
- in.readBytes(b);
- message.setPayload(b);
+// byte[] b = new byte[payloadSize];
+ ByteBuf bb = Unpooled.buffer(payloadSize);
+ in.readBytes(bb);
+ message.setPayload(bb.nioBuffer());
out.add(message);
}
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 4cebae39..b84df9f7 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
@@ -2,6 +2,8 @@ package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
+import java.nio.Buffer;
+import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
@@ -65,7 +67,8 @@ public class PublishDecoderTest {
public void testHeaderWithMessageID_Payload() throws Exception {
m_buff = Unpooled.buffer(14);
int messageID = 123;
- byte[] payload = new byte[]{0x0A, 0x0B, 0x0C};
+// byte[] payload = new byte[]{0x0A, 0x0B, 0x0C};
+ ByteBuffer payload = ByteBuffer.allocate(3).put(new byte[]{0x0A, 0x0B, 0x0C});
initHeaderWithMessageID_Payload(m_buff, messageID, payload);
//Exercise
@@ -76,7 +79,8 @@ public class PublishDecoderTest {
assertNotNull(message);
assertEquals("Fake Topic", message.getTopicName());
assertEquals(messageID, (int) message.getMessageID());
- TestUtils.verifyEquals(payload, message.getPayload());
+// TestUtils.verifyEquals(payload, message.getPayload());
+ assertEquals(payload, message.getPayload());
}
@Test
@@ -141,7 +145,9 @@ public class PublishDecoderTest {
PublishMessage message = (PublishMessage)m_results.get(0);
assertNotNull(message);
assertEquals("/topic", message.getTopicName());
- assertEquals("Test my payload", new String(message.getPayload()));
+// assertEquals("Test my payload", new String(message.getPayload()));
+ Buffer expectedPayload = ByteBuffer.allocate(15).put("Test my payload".getBytes()).flip();
+ assertEquals(expectedPayload, message.getPayload());
}
private void initHeader(ByteBuf buff) throws IllegalAccessException {
@@ -170,6 +176,16 @@ public class PublishDecoderTest {
buff.writeBytes(tmp);
}
+ private void initHeaderWithMessageID_Payload(ByteBuf buff, int messageID, ByteBuffer payload) throws IllegalAccessException {
+ ByteBuf tmp = Unpooled.buffer(4).writeBytes(Utils.encodeString("Fake Topic"));
+ tmp.writeShort(messageID);
+ tmp.writeBytes(payload);
+ buff.clear().writeByte(AbstractMessage.PUBLISH << 4 | 0x02) //set Qos to 1
+ .writeBytes(Utils.encodeRemainingLength(tmp.readableBytes()));
+ //topic name
+ buff.writeBytes(tmp);
+ }
+
private ByteBuf generatePublishQoS0(ByteBuf payload) throws IllegalAccessException {
int size = payload.capacity();
ByteBuf messageBody = Unpooled.buffer(size);
diff --git a/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/PublishEncoderTest.java b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/PublishEncoderTest.java
index a7216af1..ec0e9261 100644
--- a/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/PublishEncoderTest.java
+++ b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/PublishEncoderTest.java
@@ -3,6 +3,7 @@ package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
+import java.nio.ByteBuffer;
import org.junit.Before;
import org.junit.Test;
import static org.junit.Assert.*;
@@ -35,7 +36,8 @@ public class PublishEncoderTest {
msg.setTopicName(topic);
//variable part
- byte[] payload = new byte[]{0x0A, 0x0B, 0x0C};
+ byte[] bpayload = new byte[]{0x0A, 0x0B, 0x0C};
+ ByteBuffer payload = (ByteBuffer) ByteBuffer.allocate(3).put(bpayload).flip();
msg.setPayload(payload);
//Exercise
@@ -48,7 +50,8 @@ public class PublishEncoderTest {
//Variable part
verifyString(topic, m_out);
- verifyBuff(payload.length, payload, m_out);
+ payload.rewind();
+ verifyBuff(bpayload.length, payload, m_out);
}
@@ -61,7 +64,8 @@ public class PublishEncoderTest {
msg.setTopicName(topic);
//variable part
- byte[] payload = new byte[]{0x0A, 0x0B, 0x0C};
+ byte[] bpayload = new byte[]{0x0A, 0x0B, 0x0C};
+ ByteBuffer payload = (ByteBuffer) ByteBuffer.allocate(3).put(bpayload).flip();
msg.setPayload(payload);
//Exercise
@@ -76,7 +80,8 @@ public class PublishEncoderTest {
verifyString(topic, m_out);
assertEquals(0, m_out.readByte()); //MessageID MSB
assertEquals(1, m_out.readByte()); //MessageID LSB
- verifyBuff(payload.length, payload, m_out);
+ payload.rewind();
+ verifyBuff(bpayload.length, payload, m_out);
}
@Test(expected = IllegalArgumentException.class)
@@ -120,7 +125,8 @@ public class PublishEncoderTest {
msg.setTopicName(null);
//variable part
- byte[] payload = new byte[]{0x0A, 0x0B, 0x0C};
+ byte[] bpayload = new byte[]{0x0A, 0x0B, 0x0C};
+ ByteBuffer payload = (ByteBuffer) ByteBuffer.allocate(3).put(bpayload).flip();
msg.setPayload(payload);
ByteBuf out = Unpooled.buffer();
@@ -136,7 +142,8 @@ public class PublishEncoderTest {
msg.setTopicName(topic);
//variable part
- byte[] payload = "Test my payload".getBytes();
+ byte[] bpayload = "Test my payload".getBytes();
+ ByteBuffer payload = (ByteBuffer) ByteBuffer.allocate(bpayload.length).put(bpayload).flip();
msg.setPayload(payload);
//Exercise
@@ -145,16 +152,17 @@ public class PublishEncoderTest {
//Verify
//2 byte header + (2+6) topic name + [2 message ID if QoQ <> 0]+ payload
int size = m_out.readableBytes();
- assertEquals(10+payload.length, size);
+ assertEquals(10 + bpayload.length, size);
assertEquals(0x30, m_out.readByte()); //1 byte
assertEquals(23, m_out.readByte()); //1 byte the length
}
@Test
public void testEncodeBigMessage() throws Exception {
- byte[] payload = new byte[256];
+ byte[] bpayload = new byte[256];
- TestUtils.generateRandomPayload(256).readBytes(payload);
+ TestUtils.generateRandomPayload(256).readBytes(bpayload);
+ ByteBuffer payload = (ByteBuffer) ByteBuffer.allocate(bpayload.length).put(bpayload).flip();
String topic = "/topic";
PublishMessage msg = new PublishMessage();
@@ -168,7 +176,7 @@ public class PublishEncoderTest {
m_encoder.encode(m_mockedContext, msg, m_out);
int size = m_out.readableBytes();
- assertEquals(11+payload.length, size);
+ assertEquals(11 + bpayload.length, size);
assertEquals(0x30, m_out.readByte()); //1 byte
}
diff --git a/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/TestUtils.java b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/TestUtils.java
index 29a93411..250dc109 100644
--- a/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/TestUtils.java
+++ b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/TestUtils.java
@@ -6,6 +6,7 @@ import io.netty.buffer.Unpooled;
import io.netty.buffer.UnpooledByteBufAllocator;
import io.netty.channel.ChannelHandlerContext;
import java.io.UnsupportedEncodingException;
+import java.nio.ByteBuffer;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.*;
@@ -31,6 +32,11 @@ public class TestUtils {
}
}
+ static void verifyBuff(int numBytes, ByteBuffer bytes, ByteBuf buff) {
+ assertTrue(numBytes <= buff.readableBytes());
+ assertEquals(bytes, buff.nioBuffer());
+ }
+
/**
* Verify the presence of the given string starting from the current position
* inside the buffer.