From 3fa3705d550d947ae58d0b6b43cee5d7893dd8c4 Mon Sep 17 00:00:00 2001 From: andrea Date: Thu, 12 Sep 2013 16:50:36 +0200 Subject: [PATCH] Added encode/decode for PingReq message --- .../moquette/parser/netty/MQTTDecoder.java | 2 ++ .../moquette/parser/netty/MQTTEncoder.java | 2 ++ .../moquette/parser/netty/PingReqDecoder.java | 26 +++++++++++++++++++ .../moquette/parser/netty/PingReqEncoder.java | 19 ++++++++++++++ 4 files changed, 49 insertions(+) create mode 100644 netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PingReqDecoder.java create mode 100644 netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PingReqEncoder.java diff --git a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/MQTTDecoder.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/MQTTDecoder.java index fc5659eb..136c6b03 100644 --- a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/MQTTDecoder.java +++ b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/MQTTDecoder.java @@ -25,6 +25,8 @@ public class MQTTDecoder extends ByteToMessageDecoder { m_decoderMap.put(AbstractMessage.SUBSCRIBE, new SubscribeDecoder()); m_decoderMap.put(AbstractMessage.SUBACK, new SubAckDecoder()); m_decoderMap.put(AbstractMessage.UNSUBSCRIBE, new UnsubscribeDecoder()); + m_decoderMap.put(AbstractMessage.DISCONNECT, new DisconnectDecoder()); + m_decoderMap.put(AbstractMessage.PINGREQ, new PingReqDecoder()); } @Override 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 92b8620b..f4333cb3 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 @@ -24,6 +24,8 @@ public class MQTTEncoder extends MessageToByteEncoder { m_encoderMap.put(AbstractMessage.SUBSCRIBE, new SubscribeEncoder()); m_encoderMap.put(AbstractMessage.SUBACK, new SubAckEncoder()); m_encoderMap.put(AbstractMessage.UNSUBSCRIBE, new UnsubscribeEncoder()); + m_encoderMap.put(AbstractMessage.DISCONNECT, new DisconnectEncoder()); + m_encoderMap.put(AbstractMessage.PINGREQ, new PingReqEncoder()); } @Override diff --git a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PingReqDecoder.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PingReqDecoder.java new file mode 100644 index 00000000..9bf8ff89 --- /dev/null +++ b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PingReqDecoder.java @@ -0,0 +1,26 @@ +package org.dna.mqtt.moquette.parser.netty; + +import io.netty.buffer.ByteBuf; +import io.netty.channel.ChannelHandlerContext; +import java.util.List; +import org.dna.mqtt.moquette.parser.netty.DemuxDecoder; +import org.dna.mqtt.moquette.proto.messages.PingReqMessage; + +/** + * + * @author andrea + */ +class PingReqDecoder extends DemuxDecoder { + + @Override + void decode(ChannelHandlerContext ctx, ByteBuf in, List out) throws Exception { + //Common decoding part + in.resetReaderIndex(); + PingReqMessage message = new PingReqMessage(); + if (!decodeCommonHeader(message, in)) { + in.resetReaderIndex(); + return; + } + out.add(message); + } +} diff --git a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PingReqEncoder.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PingReqEncoder.java new file mode 100644 index 00000000..c1b7d5e7 --- /dev/null +++ b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PingReqEncoder.java @@ -0,0 +1,19 @@ +package org.dna.mqtt.moquette.parser.netty; + +import io.netty.buffer.ByteBuf; +import io.netty.channel.ChannelHandlerContext; +import org.dna.mqtt.moquette.parser.netty.DemuxEncoder; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.PingReqMessage; + +/** + * + * @author andrea + */ +class PingReqEncoder extends DemuxEncoder { + + @Override + protected void encode(ChannelHandlerContext chc, PingReqMessage msg, ByteBuf out) { + out.writeByte(AbstractMessage.PINGREQ << 4).writeByte(0); + } +}