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

Introduced hierarchy for message encoders

Browse Source
This commit is contained in:
andrea
2013-09-11 16:47:41 +02:00
parent 6afaa46cf4
commit 22fe25da32
4 changed files with 101 additions and 64 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
+70
View File
@@ -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<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 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);
}
}
netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/DemuxEncoder.java
+13
View File
@@ -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<T extends AbstractMessage> {
abstract protected void encode(ChannelHandlerContext chc, T msg, ByteBuf bb);
}
netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/MQTTEncoder.java
+16 -62
View File
@@ -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<AbstractMessage> {
@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<Byte, DemuxEncoder> m_encoderMap = new HashMap<Byte, DemuxEncoder>();
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);
}
}
netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/MQTTEncoderTest.java → netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/ConnectEncoderTest.java
+2 -2
View File
@@ -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