Added encode/decode for SubAck message
This commit is contained in:
@@ -23,6 +23,7 @@ public class MQTTDecoder extends ByteToMessageDecoder {
|
||||
m_decoderMap.put(AbstractMessage.PUBLISH, new PublishDecoder());
|
||||
m_decoderMap.put(AbstractMessage.PUBACK, new PubAckDecoder());
|
||||
m_decoderMap.put(AbstractMessage.SUBSCRIBE, new SubscribeDecoder());
|
||||
m_decoderMap.put(AbstractMessage.SUBACK, new SubAckDecoder());
|
||||
}
|
||||
|
||||
@Override
|
||||
|
||||
@@ -22,6 +22,7 @@ public class MQTTEncoder extends MessageToByteEncoder<AbstractMessage> {
|
||||
m_encoderMap.put(AbstractMessage.PUBLISH, new PublishEncoder());
|
||||
m_encoderMap.put(AbstractMessage.PUBACK, new PubAckEncoder());
|
||||
m_encoderMap.put(AbstractMessage.SUBSCRIBE, new SubscribeEncoder());
|
||||
m_encoderMap.put(AbstractMessage.SUBACK, new SubAckEncoder());
|
||||
}
|
||||
|
||||
@Override
|
||||
|
||||
@@ -0,0 +1,43 @@
|
||||
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.AbstractMessage;
|
||||
import org.dna.mqtt.moquette.proto.messages.SubAckMessage;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author andrea
|
||||
*/
|
||||
class SubAckDecoder extends DemuxDecoder {
|
||||
|
||||
@Override
|
||||
void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
|
||||
//Common decoding part
|
||||
in.resetReaderIndex();
|
||||
SubAckMessage message = new SubAckMessage();
|
||||
if (!decodeCommonHeader(message, in)) {
|
||||
in.resetReaderIndex();
|
||||
return;
|
||||
}
|
||||
int remainingLength = message.getRemainingLength();
|
||||
|
||||
//MessageID
|
||||
message.setMessageID(in.readUnsignedShort());
|
||||
remainingLength -= 2;
|
||||
|
||||
//Qos array
|
||||
if (in.readableBytes() < remainingLength ) {
|
||||
in.resetReaderIndex();
|
||||
return;
|
||||
}
|
||||
for (int i = 0; i < remainingLength; i++) {
|
||||
byte qos = in.readByte();
|
||||
message.addType(AbstractMessage.QOSType.values()[qos]);
|
||||
}
|
||||
|
||||
out.add(message);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,36 @@
|
||||
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.SubAckMessage;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author andrea
|
||||
*/
|
||||
class SubAckEncoder extends DemuxEncoder<SubAckMessage> {
|
||||
|
||||
@Override
|
||||
protected void encode(ChannelHandlerContext chc, SubAckMessage message, ByteBuf out) {
|
||||
if (message.types().isEmpty()) {
|
||||
throw new IllegalArgumentException("Found a suback message with empty topics");
|
||||
}
|
||||
|
||||
ByteBuf variableHeaderBuff = chc.alloc().buffer(4);
|
||||
variableHeaderBuff.writeShort(message.getMessageID());
|
||||
for (AbstractMessage.QOSType c : message.types()) {
|
||||
variableHeaderBuff.writeByte(c.ordinal());
|
||||
}
|
||||
|
||||
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);
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user