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

Added release operation on all ByteBuf used in encoding part

Browse Source
This commit is contained in:
andrea
2014-02-23 12:46:37 +01:00
parent 37ec617801
commit 16ccfeda1d
6 changed files with 137 additions and 106 deletions
Download Patch File Download Diff File Expand all files Collapse all files
netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/ConnectEncoder.java
+52 -46
View File
@@ -14,57 +14,63 @@ public class ConnectEncoder extends DemuxEncoder<ConnectMessage> {
@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);
}
}
netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PubAckEncoder.java
+8 -4
View File
@@ -14,10 +14,14 @@ class PubAckEncoder extends DemuxEncoder<PubAckMessage> {
@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();
}
}
}
netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PublishEncoder.java
+20 -17
View File
@@ -21,25 +21,28 @@ class PublishEncoder extends DemuxEncoder<PublishMessage> {
}
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();
}
}
}
netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/SubAckEncoder.java
+18 -12
View File
@@ -18,19 +18,25 @@ class SubAckEncoder extends DemuxEncoder<SubAckMessage> {
}
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);
}
}
netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/SubscribeEncoder.java
+20 -14
View File
@@ -22,21 +22,27 @@ class SubscribeEncoder extends DemuxEncoder<SubscribeMessage> {
}
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);
}
}
netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/UnsubscribeEncoder.java
+19 -13
View File
@@ -23,20 +23,26 @@ class UnsubscribeEncoder extends DemuxEncoder<UnsubscribeMessage> {
}
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);
}
}