From 16ccfeda1de2bb21db58c1b7425a116bda97db37 Mon Sep 17 00:00:00 2001 From: andrea Date: Sun, 23 Feb 2014 12:46:37 +0100 Subject: [PATCH] Added release operation on all ByteBuf used in encoding part --- .../moquette/parser/netty/ConnectEncoder.java | 98 ++++++++++--------- .../moquette/parser/netty/PubAckEncoder.java | 12 ++- .../moquette/parser/netty/PublishEncoder.java | 37 +++---- .../moquette/parser/netty/SubAckEncoder.java | 30 +++--- .../parser/netty/SubscribeEncoder.java | 34 ++++--- .../parser/netty/UnsubscribeEncoder.java | 32 +++--- 6 files changed, 137 insertions(+), 106 deletions(-) diff --git a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/ConnectEncoder.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/ConnectEncoder.java index 030b32bf..a012eb75 100644 --- a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/ConnectEncoder.java +++ b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/ConnectEncoder.java @@ -14,57 +14,63 @@ public class ConnectEncoder extends DemuxEncoder { @Override protected void encode(ChannelHandlerContext chc, ConnectMessage message, ByteBuf out) { ByteBuf staticHeaderBuff = chc.alloc().buffer(12); - staticHeaderBuff.writeBytes(Utils.encodeString("MQIsdp")); - - //version - staticHeaderBuff.writeByte(0x03); - - //connection flags and Strings - byte connectionFlags = 0; - if (message.isCleanSession()) { - connectionFlags |= 0x02; - } - if (message.isWillFlag()) { - connectionFlags |= 0x04; - } - connectionFlags |= ((message.getWillQos() & 0x03) << 3); - if (message.isWillRetain()) { - connectionFlags |= 0x020; - } - if (message.isPasswordFlag()) { - connectionFlags |= 0x040; - } - if (message.isUserFlag()) { - connectionFlags |= 0x080; - } - staticHeaderBuff.writeByte(connectionFlags); - - //Keep alive timer - staticHeaderBuff.writeShort(message.getKeepAlive()); - - //Variable part + ByteBuf buff = chc.alloc().buffer(); ByteBuf variableHeaderBuff = chc.alloc().buffer(12); - if (message.getClientID() != null) { - variableHeaderBuff.writeBytes(Utils.encodeString(message.getClientID())); - if (message.isWillFlag()) { - variableHeaderBuff.writeBytes(Utils.encodeString(message.getWillTopic())); - variableHeaderBuff.writeBytes(Utils.encodeString(message.getWillMessage())); + try { + staticHeaderBuff.writeBytes(Utils.encodeString("MQIsdp")); + + //version + staticHeaderBuff.writeByte(0x03); + + //connection flags and Strings + byte connectionFlags = 0; + if (message.isCleanSession()) { + connectionFlags |= 0x02; } - if (message.isUserFlag() && message.getUsername() != null) { - variableHeaderBuff.writeBytes(Utils.encodeString(message.getUsername())); - if (message.isPasswordFlag() && message.getPassword() != null) { - variableHeaderBuff.writeBytes(Utils.encodeString(message.getPassword())); + if (message.isWillFlag()) { + connectionFlags |= 0x04; + } + connectionFlags |= ((message.getWillQos() & 0x03) << 3); + if (message.isWillRetain()) { + connectionFlags |= 0x020; + } + if (message.isPasswordFlag()) { + connectionFlags |= 0x040; + } + if (message.isUserFlag()) { + connectionFlags |= 0x080; + } + staticHeaderBuff.writeByte(connectionFlags); + + //Keep alive timer + staticHeaderBuff.writeShort(message.getKeepAlive()); + + //Variable part + if (message.getClientID() != null) { + variableHeaderBuff.writeBytes(Utils.encodeString(message.getClientID())); + if (message.isWillFlag()) { + variableHeaderBuff.writeBytes(Utils.encodeString(message.getWillTopic())); + variableHeaderBuff.writeBytes(Utils.encodeString(message.getWillMessage())); + } + if (message.isUserFlag() && message.getUsername() != null) { + variableHeaderBuff.writeBytes(Utils.encodeString(message.getUsername())); + if (message.isPasswordFlag() && message.getPassword() != null) { + variableHeaderBuff.writeBytes(Utils.encodeString(message.getPassword())); + } } } + + int variableHeaderSize = variableHeaderBuff.readableBytes(); + buff.writeByte(AbstractMessage.CONNECT << 4); + buff.writeBytes(Utils.encodeRemainingLength(12 + variableHeaderSize)); + buff.writeBytes(staticHeaderBuff).writeBytes(variableHeaderBuff); + + out.writeBytes(buff); + } finally { + staticHeaderBuff.release(); + buff.release(); + variableHeaderBuff.release(); } - - int variableHeaderSize = variableHeaderBuff.readableBytes(); - ByteBuf buff = chc.alloc().buffer(); - buff.writeByte(AbstractMessage.CONNECT << 4); - buff.writeBytes(Utils.encodeRemainingLength(12 + variableHeaderSize)); - buff.writeBytes(staticHeaderBuff).writeBytes(variableHeaderBuff); - - out.writeBytes(buff); } } diff --git a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PubAckEncoder.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PubAckEncoder.java index 56d608a4..ac3e2c26 100644 --- a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PubAckEncoder.java +++ b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PubAckEncoder.java @@ -14,10 +14,14 @@ class PubAckEncoder extends DemuxEncoder { @Override protected void encode(ChannelHandlerContext chc, PubAckMessage msg, ByteBuf out) { ByteBuf buff = chc.alloc().buffer(4); - buff.writeByte(AbstractMessage.PUBACK << 4); - buff.writeBytes(Utils.encodeRemainingLength(2)); - buff.writeShort(msg.getMessageID()); - out.writeBytes(buff); + try { + buff.writeByte(AbstractMessage.PUBACK << 4); + buff.writeBytes(Utils.encodeRemainingLength(2)); + buff.writeShort(msg.getMessageID()); + out.writeBytes(buff); + } finally { + buff.release(); + } } } diff --git a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PublishEncoder.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PublishEncoder.java index 171b4dcb..5ecd533e 100644 --- a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PublishEncoder.java +++ b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PublishEncoder.java @@ -21,25 +21,28 @@ class PublishEncoder extends DemuxEncoder { } ByteBuf variableHeaderBuff = ctx.alloc().buffer(2); - variableHeaderBuff.writeBytes(Utils.encodeString(message.getTopicName())); - if (message.getQos() == AbstractMessage.QOSType.LEAST_ONE || - message.getQos() == AbstractMessage.QOSType.EXACTLY_ONCE ) { - if (message.getMessageID() == null) { - throw new IllegalArgumentException("Found a message with QOS 1 or 2 and not MessageID setted"); + try { + variableHeaderBuff.writeBytes(Utils.encodeString(message.getTopicName())); + if (message.getQos() == AbstractMessage.QOSType.LEAST_ONE || + message.getQos() == AbstractMessage.QOSType.EXACTLY_ONCE ) { + if (message.getMessageID() == null) { + throw new IllegalArgumentException("Found a message with QOS 1 or 2 and not MessageID setted"); + } + variableHeaderBuff.writeShort(message.getMessageID()); } - variableHeaderBuff.writeShort(message.getMessageID()); - } - variableHeaderBuff.writeBytes(message.getPayload()); - int variableHeaderSize = variableHeaderBuff.readableBytes(); - - byte flags = Utils.encodeFlags(message); - - ByteBuf buff = ctx.alloc().buffer(2 + variableHeaderSize); - buff.writeByte(AbstractMessage.PUBLISH << 4 | flags); - buff.writeBytes(Utils.encodeRemainingLength(variableHeaderSize)); - buff.writeBytes(variableHeaderBuff); + variableHeaderBuff.writeBytes(message.getPayload()); + int variableHeaderSize = variableHeaderBuff.readableBytes(); - out.writeBytes(buff); + byte flags = Utils.encodeFlags(message); + + ByteBuf buff = ctx.alloc().buffer(2 + variableHeaderSize); + buff.writeByte(AbstractMessage.PUBLISH << 4 | flags); + buff.writeBytes(Utils.encodeRemainingLength(variableHeaderSize)); + buff.writeBytes(variableHeaderBuff); + out.writeBytes(buff); + } finally { + variableHeaderBuff.release(); + } } } diff --git a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/SubAckEncoder.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/SubAckEncoder.java index 6cb0dd48..89db454b 100644 --- a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/SubAckEncoder.java +++ b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/SubAckEncoder.java @@ -18,19 +18,25 @@ class SubAckEncoder extends DemuxEncoder { } ByteBuf variableHeaderBuff = chc.alloc().buffer(4); - variableHeaderBuff.writeShort(message.getMessageID()); - for (AbstractMessage.QOSType c : message.types()) { - variableHeaderBuff.writeByte(c.ordinal()); + ByteBuf buff = null; + try { + variableHeaderBuff.writeShort(message.getMessageID()); + for (AbstractMessage.QOSType c : message.types()) { + variableHeaderBuff.writeByte(c.ordinal()); + } + + int variableHeaderSize = variableHeaderBuff.readableBytes(); + buff = chc.alloc().buffer(2 + variableHeaderSize); + + buff.writeByte(AbstractMessage.SUBACK << 4 ); + buff.writeBytes(Utils.encodeRemainingLength(variableHeaderSize)); + buff.writeBytes(variableHeaderBuff); + + out.writeBytes(buff); + } finally { + variableHeaderBuff.release(); + buff.release(); } - - int variableHeaderSize = variableHeaderBuff.readableBytes(); - ByteBuf buff = chc.alloc().buffer(2 + variableHeaderSize); - - buff.writeByte(AbstractMessage.SUBACK << 4 ); - buff.writeBytes(Utils.encodeRemainingLength(variableHeaderSize)); - buff.writeBytes(variableHeaderBuff); - - out.writeBytes(buff); } } diff --git a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/SubscribeEncoder.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/SubscribeEncoder.java index c97749f7..694fa2da 100644 --- a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/SubscribeEncoder.java +++ b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/SubscribeEncoder.java @@ -22,21 +22,27 @@ class SubscribeEncoder extends DemuxEncoder { } ByteBuf variableHeaderBuff = chc.alloc().buffer(4); - variableHeaderBuff.writeShort(message.getMessageID()); - for (SubscribeMessage.Couple c : message.subscriptions()) { - variableHeaderBuff.writeBytes(Utils.encodeString(c.getTopic())); - variableHeaderBuff.writeByte(c.getQos()); + ByteBuf buff = null; + try { + variableHeaderBuff.writeShort(message.getMessageID()); + for (SubscribeMessage.Couple c : message.subscriptions()) { + variableHeaderBuff.writeBytes(Utils.encodeString(c.getTopic())); + variableHeaderBuff.writeByte(c.getQos()); + } + + int variableHeaderSize = variableHeaderBuff.readableBytes(); + byte flags = Utils.encodeFlags(message); + buff = chc.alloc().buffer(2 + variableHeaderSize); + + buff.writeByte(AbstractMessage.SUBSCRIBE << 4 | flags); + buff.writeBytes(Utils.encodeRemainingLength(variableHeaderSize)); + buff.writeBytes(variableHeaderBuff); + + out.writeBytes(buff); + } finally { + variableHeaderBuff.release(); + buff.release(); } - - int variableHeaderSize = variableHeaderBuff.readableBytes(); - byte flags = Utils.encodeFlags(message); - ByteBuf buff = chc.alloc().buffer(2 + variableHeaderSize); - - buff.writeByte(AbstractMessage.SUBSCRIBE << 4 | flags); - buff.writeBytes(Utils.encodeRemainingLength(variableHeaderSize)); - buff.writeBytes(variableHeaderBuff); - - out.writeBytes(buff); } } diff --git a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/UnsubscribeEncoder.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/UnsubscribeEncoder.java index 3800264a..07827e36 100644 --- a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/UnsubscribeEncoder.java +++ b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/UnsubscribeEncoder.java @@ -23,20 +23,26 @@ class UnsubscribeEncoder extends DemuxEncoder { } ByteBuf variableHeaderBuff = chc.alloc().buffer(4); - variableHeaderBuff.writeShort(message.getMessageID()); - for (String topic : message.topics()) { - variableHeaderBuff.writeBytes(Utils.encodeString(topic)); + ByteBuf buff = null; + try { + variableHeaderBuff.writeShort(message.getMessageID()); + for (String topic : message.topics()) { + variableHeaderBuff.writeBytes(Utils.encodeString(topic)); + } + + int variableHeaderSize = variableHeaderBuff.readableBytes(); + byte flags = Utils.encodeFlags(message); + buff = chc.alloc().buffer(2 + variableHeaderSize); + + buff.writeByte(AbstractMessage.UNSUBSCRIBE << 4 | flags); + buff.writeBytes(Utils.encodeRemainingLength(variableHeaderSize)); + buff.writeBytes(variableHeaderBuff); + + out.writeBytes(buff); + } finally { + variableHeaderBuff.release(); + buff.release(); } - - int variableHeaderSize = variableHeaderBuff.readableBytes(); - byte flags = Utils.encodeFlags(message); - ByteBuf buff = chc.alloc().buffer(2 + variableHeaderSize); - - buff.writeByte(AbstractMessage.UNSUBSCRIBE << 4 | flags); - buff.writeBytes(Utils.encodeRemainingLength(variableHeaderSize)); - buff.writeBytes(variableHeaderBuff); - - out.writeBytes(buff); } }