From 22fe25da3287bf9bc33a532fbdedf70767f7046c Mon Sep 17 00:00:00 2001 From: andrea Date: Wed, 11 Sep 2013 16:47:41 +0200 Subject: [PATCH] Introduced hierarchy for message encoders --- .../moquette/parser/netty/ConnectEncoder.java | 70 +++++++++++++++++ .../moquette/parser/netty/DemuxEncoder.java | 13 ++++ .../moquette/parser/netty/MQTTEncoder.java | 78 ++++--------------- ...coderTest.java => ConnectEncoderTest.java} | 4 +- 4 files changed, 101 insertions(+), 64 deletions(-) create mode 100644 netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/ConnectEncoder.java create mode 100644 netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/DemuxEncoder.java rename netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/{MQTTEncoderTest.java => ConnectEncoderTest.java} (98%) 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 new file mode 100644 index 00000000..030b32bf --- /dev/null +++ b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/ConnectEncoder.java @@ -0,0 +1,70 @@ +package org.dna.mqtt.moquette.parser.netty; + +import io.netty.buffer.ByteBuf; +import io.netty.channel.ChannelHandlerContext; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.ConnectMessage; + +/** + * + * @author andrea + */ +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 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())); + } + 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(); + 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/DemuxEncoder.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/DemuxEncoder.java new file mode 100644 index 00000000..9499dd82 --- /dev/null +++ b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/DemuxEncoder.java @@ -0,0 +1,13 @@ +package org.dna.mqtt.moquette.parser.netty; + +import io.netty.buffer.ByteBuf; +import io.netty.channel.ChannelHandlerContext; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; + +/** + * + * @author andrea + */ +abstract class DemuxEncoder { + abstract protected void encode(ChannelHandlerContext chc, T msg, ByteBuf bb); +} diff --git a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/MQTTEncoder.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/MQTTEncoder.java index c89aefb1..50f4062b 100644 --- a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/MQTTEncoder.java +++ b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/MQTTEncoder.java @@ -2,76 +2,30 @@ package org.dna.mqtt.moquette.parser.netty; import io.netty.buffer.ByteBuf; import io.netty.channel.ChannelHandlerContext; +import io.netty.handler.codec.CorruptedFrameException; import io.netty.handler.codec.MessageToByteEncoder; +import java.util.HashMap; +import java.util.Map; import org.dna.mqtt.moquette.proto.messages.AbstractMessage; -import org.dna.mqtt.moquette.proto.messages.ConnectMessage; /** * * @author andrea */ public class MQTTEncoder extends MessageToByteEncoder { - - @Override - protected void encode(ChannelHandlerContext chc, AbstractMessage msg, ByteBuf bb) throws Exception { - if (msg.getMessageType() == AbstractMessage.CONNECT) { - encodeConnect(chc, (ConnectMessage)msg, bb); - } - } - - private void encodeConnect(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 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())); - } - 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(); - ByteBuf buff = chc.alloc().buffer(); - buff.writeByte(AbstractMessage.CONNECT << 4); - buff.writeBytes(Utils.encodeRemainingLength(12 + variableHeaderSize)); - buff.writeBytes(staticHeaderBuff).writeBytes(variableHeaderBuff); - - out.writeBytes(buff); + + private Map m_encoderMap = new HashMap(); + + MQTTEncoder() { + m_encoderMap.put(AbstractMessage.CONNECT, new ConnectEncoder()); } + @Override + protected void encode(ChannelHandlerContext chc, AbstractMessage msg, ByteBuf bb) throws Exception { + DemuxEncoder encoder = m_encoderMap.get(msg.getMessageType()); + if (encoder == null) { + throw new CorruptedFrameException("Can't find any suitable decoder for message type: " + msg.getMessageType()); + } + encoder.encode(chc, msg, bb); + } } diff --git a/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/MQTTEncoderTest.java b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/ConnectEncoderTest.java similarity index 98% rename from netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/MQTTEncoderTest.java rename to netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/ConnectEncoderTest.java index 6fe50536..8bbb3a3c 100644 --- a/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/MQTTEncoderTest.java +++ b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/ConnectEncoderTest.java @@ -17,8 +17,8 @@ import static org.dna.mqtt.moquette.parser.netty.TestUtils.*; * * @author andrea */ -public class MQTTEncoderTest { - MQTTEncoder m_encoder = new MQTTEncoder(); +public class ConnectEncoderTest { + ConnectEncoder m_encoder = new ConnectEncoder(); ChannelHandlerContext m_mockedContext; @Before