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 136c6b03..79d74027 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 @@ -27,6 +27,7 @@ public class MQTTDecoder extends ByteToMessageDecoder { m_decoderMap.put(AbstractMessage.UNSUBSCRIBE, new UnsubscribeDecoder()); m_decoderMap.put(AbstractMessage.DISCONNECT, new DisconnectDecoder()); m_decoderMap.put(AbstractMessage.PINGREQ, new PingReqDecoder()); + m_decoderMap.put(AbstractMessage.PINGRESP, new PingRespDecoder()); } @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 f4333cb3..aea7e3d6 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 @@ -26,6 +26,7 @@ public class MQTTEncoder extends MessageToByteEncoder { m_encoderMap.put(AbstractMessage.UNSUBSCRIBE, new UnsubscribeEncoder()); m_encoderMap.put(AbstractMessage.DISCONNECT, new DisconnectEncoder()); m_encoderMap.put(AbstractMessage.PINGREQ, new PingReqEncoder()); + m_encoderMap.put(AbstractMessage.PINGRESP, new PingRespEncoder()); } @Override diff --git a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PingRespDecoder.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PingRespDecoder.java new file mode 100644 index 00000000..35d3b313 --- /dev/null +++ b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PingRespDecoder.java @@ -0,0 +1,25 @@ +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.proto.messages.PingRespMessage; + +/** + * + * @author andrea + */ +class PingRespDecoder extends DemuxDecoder { + + @Override + void decode(ChannelHandlerContext ctx, ByteBuf in, List out) throws Exception { + //Common decoding part + in.resetReaderIndex(); + PingRespMessage message = new PingRespMessage(); + if (!decodeCommonHeader(message, in)) { + in.resetReaderIndex(); + return; + } + out.add(message); + } +} diff --git a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PingRespEncoder.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PingRespEncoder.java new file mode 100644 index 00000000..16833214 --- /dev/null +++ b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PingRespEncoder.java @@ -0,0 +1,18 @@ +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.PingRespMessage; + +/** + * + * @author andrea + */ +class PingRespEncoder extends DemuxEncoder { + + @Override + protected void encode(ChannelHandlerContext chc, PingRespMessage msg, ByteBuf out) { + out.writeByte(AbstractMessage.PINGRESP << 4).writeByte(0); + } +}