public class ConnAckMessage extends Message {
	
	public enum ConnectionStatus {
		ACCEPTED(0x00),

		UNACCEPTABLE_PROTOCOL_VERSION(0x01),

		IDENTIFIER_REJECTED(0x02),

		SERVER_UNAVAILABLE(0x03),

		BAD_USERNAME_OR_PASSWORD(0x04),

		NOT_AUTHORIZED(0x05);
		
		private final int value;
		
		ConnectionStatus(int value) {
			this.value = value;
		}
		
		/**
		 * 获取类型对应的值
		 * @return int
		 * @author zer0
		 * @version 1.0
		 * @date 2016-3-4
		 */
		public int value() {
			return value;
		}
		
		//通过读取到的整型来获取对应的QoS类型
		public static ConnectionStatus valueOf(byte i) {
			for(ConnectionStatus q: ConnectionStatus.values()) {
				if (q.value == i)
					return q;
			}
			throw new IllegalArgumentException("连接响应值无效: " + i);
		}
	}
	
	
	public ConnAckMessage(FixedHeader fixedHeader, ConnAckVariableHeader variableHeader) {
		super(fixedHeader, variableHeader);
	}
	
	@Override
	public ConnAckVariableHeader getVariableHeader(){
		return (ConnAckVariableHeader)super.getVariableHeader();
	}

} public class ConnectMessage extends Message {
	
	public ConnectMessage(FixedHeader fixedHeader, ConnectVariableHeader variableHeader,
			ConnectPayload payload) {
		super(fixedHeader, variableHeader, payload);
	}
	
	@Override
	public ConnectVariableHeader getVariableHeader() {
		return (ConnectVariableHeader)super.getVariableHeader();
	}
	
	@Override
	public ConnectPayload getPayload() {
		return (ConnectPayload)super.getPayload();
	}

} public class ConnectPayload{
	private String clientId;//客户端ID
	private String willTopic;
	private String willMessage;
	private String username;//如果设置User Name标识，可以在此读取用户名称
	private String password;//如果设置Password标识，便可读取用户密码
	
	public ConnectPayload(String clientId, String willTopic,
			String willMessage, String username, String password) {
		super();
		this.clientId = clientId;
		this.willTopic = willTopic;
		this.willMessage = willMessage;
		this.username = username;
		this.password = password;
	}
	
	public String getClientId() {
		return clientId;
	}
	public void setClientId(String clientId) {
		this.clientId = clientId;
	}
	public String getWillTopic() {
		return willTopic;
	}
	public void setWillTopic(String willTopic) {
		this.willTopic = willTopic;
	}
	public String getWillMessage() {
		return willMessage;
	}
	public void setWillMessage(String willMessage) {
		this.willMessage = willMessage;
	}
	public String getUsername() {
		return username;
	}
	public void setUsername(String username) {
		this.username = username;
	}
	public String getPassword() {
		return password;
	}
	public void setPassword(String password) {
		this.password = password;
	}
	
} public class ConnectVariableHeader{
	private String protocolName = "MQTT";//协议规定的协议名
	private byte protocolVersionNumber = 4;//MQTT_v3.1.1协议的版本号
	
	//Connect Flags的六个参数
	private boolean hasUsername;//是否有用户名，与密码一起，要么都为0，要么都为1，否则无效
	private boolean hasPassword;//是否有密码，与用户名一起，要么都为0，要么都为1，否则无效
	private boolean willRetain;//设置Will Flag为1，Will Retain标志就是有效的，当客户端意
							   //外断开服务器发布其Will Message之后，服务器是否应该继续保存
	private QoS willQoS;//设置Will Flag为1，Will QoS标志就是有效的
	private boolean hasWill;//是否设置遗嘱，设置以后，遗嘱生效。遗嘱就是客户端预先定义好，在自己
	                        //异常断开的情况下，所留下的最后遗愿
	private boolean cleanSession;//是否清理session
	private boolean reservedIsZero;//协议的保留位，此位必须校验且必须为0，不为0则断开连接
	
	private int keepAlive;//心跳包时长

	public ConnectVariableHeader(String protocolName,
			byte protocolVersionNumber, boolean hasUsername,
			boolean hasPassword, boolean willRetain, QoS willQoS,
			boolean hasWill, boolean cleanSession, boolean reservedIsZero,
			int keepAlive) {
		this.protocolName = protocolName;
		this.protocolVersionNumber = protocolVersionNumber;
		this.hasUsername = hasUsername;
		this.hasPassword = hasPassword;
		this.willRetain = willRetain;
		this.willQoS = willQoS;
		this.hasWill = hasWill;
		this.cleanSession = cleanSession;
		this.reservedIsZero = reservedIsZero;
		this.keepAlive = keepAlive;
	}

	public String getProtocolName() {
		return protocolName;
	}

	public byte getProtocolVersionNumber() {
		return protocolVersionNumber;
	}

	public boolean isHasUsername() {
		return hasUsername;
	}

	public boolean isHasPassword() {
		return hasPassword;
	}
	
	public boolean isWillRetain() {
		return willRetain;
	}

	public QoS getWillQoS() {
		return willQoS;
	}

	public boolean isHasWill() {
		return hasWill;
	}
	
	public boolean isCleanSession() {
		return cleanSession;
	}

	public boolean isReservedIsZero() {
		return reservedIsZero;
	}

	public int getKeepAlive() {
		return keepAlive;
	}
	
} public class FixedHeader {
	private final MessageType messageType;
	private boolean dup; //MQTT协议头第5bit，代表打开标志，表示是否第一次发送
	private QoS qos; //MQTT协议头前6,7bit，代表服务质量
	private boolean retain; //MQTT协议头第8bit，代表是否保持
	private int messageLength; //第二个字节
	
	public FixedHeader(MessageType messageType,
			   boolean dup,
			   QoS qos,
			   boolean retain){
		this.messageType = messageType;
		this.dup = dup;
		this.qos = qos;
		this.retain = retain;
	}
	
	public FixedHeader(MessageType messageType,
					   boolean dup,
					   QoS qos,
					   boolean retain,
					   int messageLength){
		this.messageType = messageType;
		this.dup = dup;
		this.qos = qos;
		this.retain = retain;
		this.messageLength = messageLength;
	}

	public boolean isDup() {
		return dup;
	}

	public void setDup(boolean dup) {
		this.dup = dup;
	}

	public QoS getQos() {
		return qos;
	}

	public void setQos(QoS qos) {
		this.qos = qos;
	}

	public boolean isRetain() {
		return retain;
	}

	public void setRetain(boolean retain) {
		this.retain = retain;
	}

	public int getMessageLength() {
		return messageLength;
	}

	public void setMessageLength(int messageLength) {
		this.messageLength = messageLength;
	}

	public MessageType getMessageType() {
		return messageType;
	}
	
	@Override
	public String toString() {
		StringBuilder stringBuilder = new StringBuilder();
		stringBuilder.append(StringUtil.simpleClassName(this))
					 .append('[')
					 .append("messageType=").append(messageType)
					 .append(", isDup=").append(dup)
					 .append(", qosLevel=").append(qos)
					 .append(", isRetain=").append(retain)
					 .append(", messageLength=").append(messageLength)
					 .append(']');
		return stringBuilder.toString();
	}
	
	public static FixedHeader getConnectFixedHeader(){
		return new FixedHeader(MessageType.CONNECT, false, QoS.AT_MOST_ONCE, false);
	}
	
	public static FixedHeader getConnAckFixedHeader(){
		return new FixedHeader(MessageType.CONNACK, false, QoS.AT_MOST_ONCE, false);
	}
	
	public static FixedHeader getPublishFixedHeader(
			   boolean dup,
			   QoS qos,
			   boolean retain){
		return new FixedHeader(MessageType.PUBLISH, dup, qos, retain);
	}
	
	public static FixedHeader getPubAckFixedHeader(){
		return new FixedHeader(MessageType.PUBACK, false, QoS.AT_MOST_ONCE, false);
	}
	
	public static FixedHeader getPubRecFixedHeader(){
		return new FixedHeader(MessageType.PUBREC, false, QoS.AT_MOST_ONCE, false);
	}
	
	public static FixedHeader getPubRelFixedHeader(){
		return new FixedHeader(MessageType.PUBREL, false, QoS.AT_LEAST_ONCE, false);
	}
	
	public static FixedHeader getPubCompFixedHeader(){
		return new FixedHeader(MessageType.PUBCOMP, false, QoS.AT_MOST_ONCE, false);
	}
	
	public static FixedHeader getSubscribeFixedHeader(){
		return new FixedHeader(MessageType.SUBSCRIBE, false, QoS.AT_LEAST_ONCE, false);
	}
	
	public static FixedHeader getSubAckFixedHeader(){
		return new FixedHeader(MessageType.SUBACK, false, QoS.AT_MOST_ONCE, false);
	}
	
	public static FixedHeader getUnSubscribeFixedHeader(){
		return new FixedHeader(MessageType.UNSUBSCRIBE, false, QoS.AT_LEAST_ONCE, false);
	}
	
	public static FixedHeader getUnSubAckFixedHeader(){
		return new FixedHeader(MessageType.UNSUBACK, false, QoS.AT_MOST_ONCE, false);
	}
	
	public static FixedHeader getPingRespFixedHeader(){
		return new FixedHeader(MessageType.PINGRESP, false, QoS.AT_MOST_ONCE, false);
	}
	
	public static FixedHeader getDisconnectFixedHeader(){
		return new FixedHeader(MessageType.DISCONNECT, false, QoS.AT_MOST_ONCE, false);
	}
} public class Message {
	
	private final FixedHeader fixedHeader;
	private final Object variableHeader;
	private final Object payload;
	
	public Message(FixedHeader fixedHeader,
				   Object variableHeader,
				   Object payload){
		this.fixedHeader = fixedHeader;
		this.variableHeader = variableHeader;
		this.payload = payload;
	}
	
	/**
	 * 初始化message，针对没有可变头部和荷载的协议类型
	 * @author zer0
	 * @version 1.0
	 * @date 2016-3-3
	 */
	public Message(FixedHeader fixedHeader){
		this(fixedHeader, null, null);
	}
	
	/**
	 * 初始化message，针对没有荷载的协议类型
	 * @author zer0
	 * @version 1.0
	 * @date 2016-3-3
	 */
	public Message(FixedHeader fixedHeader, Object variableHeader){
		this(fixedHeader, variableHeader, null);
	}

	public FixedHeader getFixedHeader() {
		return fixedHeader;
	}

	public Object getVariableHeader() {
		return variableHeader;
	}

	public Object getPayload() {
		return payload;
	}
	
	@Override
	public String toString() {
		StringBuilder stringBuilder = new StringBuilder();
		stringBuilder.append(StringUtil.simpleClassName(this))
					 .append("[")
					 .append("fixedHeader=")
					 .append(getFixedHeader() != null?getFixedHeader().toString():"")
					 .append(",variableHeader=")
					 .append(getVariableHeader() != null?getVariableHeader().toString():"")
					 .append(",payload=")
					 .append(getPayload() != null?getPayload().toString():"")
					 .append("]");
		return stringBuilder.toString();
	}

} public class PackageIdVariableHeader {
	
	private int packageID;

	public PackageIdVariableHeader(int packageID) {
		if (packageID < 1 || packageID > 65535) {
			throw new IllegalArgumentException("消息ID:" + packageID + "必须在1~65535范围内");
		}
		this.packageID = packageID;
	}

	public int getPackageID() {
		return packageID;
	}

	public void setPackageID(int packageID) {
		this.packageID = packageID;
	}
	
} public class PublishMessage extends Message {

	public PublishMessage(FixedHeader fixedHeader, PublishVariableHeader variableHeader,
			ByteBuf payload) {
		super(fixedHeader, variableHeader, payload);
	}
	
	@Override
	public PublishVariableHeader getVariableHeader() {
		return (PublishVariableHeader)super.getVariableHeader();
	}
	
	@Override
	public ByteBuf getPayload() {
		return (ByteBuf)super.getPayload();
	}

} public class PublishVariableHeader{
	private String topic;
	private int packageID;
	
	public PublishVariableHeader(String topic) {
		this.topic = topic;
	}
	
	public PublishVariableHeader(String topic, int packageID) {
		this.topic = topic;
		this.packageID = packageID;
	}
	
	public String getTopic() {
		return topic;
	}
	public void setTopic(String topic) {
		this.topic = topic;
	}
	public int getPackageID() {
		return packageID;
	}
	public void setPackageID(int packageID) {
		this.packageID = packageID;
	}
	
} public class SubAckMessage extends Message {

	public SubAckMessage(FixedHeader fixedHeader, PackageIdVariableHeader variableHeader,
			SubAckPayload payload) {
		super(fixedHeader, variableHeader, payload);
	}
	
	@Override
	public PackageIdVariableHeader getVariableHeader() {
		return (PackageIdVariableHeader)super.getVariableHeader();
	}
	
	@Override
	public SubAckPayload getPayload() {
		return (SubAckPayload)super.getPayload();
	}
} public class SubAckPayload{
	private List<Integer> grantedQosLevel;
	
	public SubAckPayload(List<Integer> grantedQosLevel) {
		this.grantedQosLevel = grantedQosLevel;
	}

	public List<Integer> getGrantedQosLevel() {
		return grantedQosLevel;
	}
	
} public class SubscribeMessage extends Message {

	public SubscribeMessage(FixedHeader fixedHeader, PackageIdVariableHeader variableHeader,
			SubscribePayload payload) {
		super(fixedHeader, variableHeader, payload);
	}
	
	@Override
	public PackageIdVariableHeader getVariableHeader() {
		return (PackageIdVariableHeader)super.getVariableHeader();
	}
	
	@Override
	public SubscribePayload getPayload() {
		return (SubscribePayload)super.getPayload();
	}
	
} public class SubscribePayload{
	private List<TopicSubscribe> topicSubscribes;

	public SubscribePayload(List<TopicSubscribe> topicSubscribes) {
		super();
		this.topicSubscribes = topicSubscribes;
	}

	public List<TopicSubscribe> getTopicSubscribes() {
		return topicSubscribes;
	}

} public class TopicSubscribe{
	private String topicFilter;
	private QoS qos;
	
	public TopicSubscribe(String topicFilter, QoS qos) {
		super();
		this.topicFilter = topicFilter;
		this.qos = qos;
	}
	
	public String getTopicFilter() {
		return topicFilter;
	}
	public void setTopicFilter(String topicFilter) {
		this.topicFilter = topicFilter;
	}
	public QoS getQos() {
		return qos;
	}
	public void setQos(QoS qos) {
		this.qos = qos;
	}
} public class UnSubscribeMessage extends Message {

	public UnSubscribeMessage(FixedHeader fixedHeader, PackageIdVariableHeader variableHeader,
			UnSubscribePayload payload) {
		super(fixedHeader, variableHeader, payload);
	}
	
	@Override
	public PackageIdVariableHeader getVariableHeader() {
		return (PackageIdVariableHeader)super.getVariableHeader();
	}
	
	@Override
	public UnSubscribePayload getPayload() {
		return (UnSubscribePayload)super.getPayload();
	}
	
} public class UnSubscribePayload{
	
	private List<String> topics;
	
	public UnSubscribePayload(List<String> topics) {
		super();
		this.topics = topics;
	}

	public List<String> getTopics() {
		return topics;
	}

}
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.CorruptedFrameException;
import io.netty.handler.codec.MessageToByteEncoder;
import io.netty.util.CharsetUtil;
import java.util.List;
import org.apache.log4j.Logger;
import com.syxy.protocol.mqttImp.message.ConnAckMessage;
import com.syxy.protocol.mqttImp.message.ConnectMessage;
import com.syxy.protocol.mqttImp.message.ConnectPayload;
import com.syxy.protocol.mqttImp.message.ConnectVariableHeader;
import com.syxy.protocol.mqttImp.message.FixedHeader;
import com.syxy.protocol.mqttImp.message.Message;
import com.syxy.protocol.mqttImp.message.PackageIdVariableHeader;
import com.syxy.protocol.mqttImp.message.PublishMessage;
import com.syxy.protocol.mqttImp.message.PublishVariableHeader;
import com.syxy.protocol.mqttImp.message.SubAckMessage;
import com.syxy.protocol.mqttImp.message.SubAckPayload;
import com.syxy.protocol.mqttImp.message.SubscribeMessage;
import com.syxy.protocol.mqttImp.message.SubscribePayload;
import com.syxy.protocol.mqttImp.message.TopicSubscribe;
import com.syxy.protocol.mqttImp.message.UnSubscribeMessage;
import com.syxy.protocol.mqttImp.message.UnSubscribePayload;
package com.syxy.protocol.mqttImp;





/**
 *  MQTT协议编码
 *  修改（1）
 * 
 * @author zer0
 * @version 1.0
 * @date 2015-2-16
 * @date 2016-3-3(1)
 */
public class MQTTEncoder extends MessageToByteEncoder<Message> {
	
	private final static Logger Log = Logger.getLogger(MQTTEncoder.class);
	public static final int MAX_LENGTH_LIMIT = 268435455;
	private final byte[] EMPTY = new byte[0];
	private final int UTF8_FIX_LENGTH = 2;//UTF编码的byte，最开始必须为2字节的长度字段
	
	@Override
	protected void encode(ChannelHandlerContext ctx, Message msg, ByteBuf out)
			throws Exception {
		ByteBufAllocator byteBufAllocator = ctx.alloc();
		ByteBuf encodedByteBuf;
		
		switch (msg.getFixedHeader().getMessageType()) {
		case CONNECT:
			encodedByteBuf = encodeConnectMessage(byteBufAllocator, (ConnectMessage)msg);
			break;
		case CONNACK:
			encodedByteBuf = encodeConnAckMessage(byteBufAllocator, (ConnAckMessage)msg);
			break;
		case PUBLISH:
			encodedByteBuf = encodePublishMessage(byteBufAllocator, (PublishMessage)msg);
			break;
		case SUBSCRIBE:
			encodedByteBuf = encodeSubscribeMessage(byteBufAllocator, (SubscribeMessage)msg);