public final class MQTTMesageFactory {

    public static Message newMessage(FixedHeader fixedHeader, Object variableHeader, Object payload) {
        switch (fixedHeader.getMessageType()) {
            case CONNECT :
                return new ConnectMessage(fixedHeader, 
                		(ConnectVariableHeader)variableHeader, 
                		(ConnectPayload)payload);

            case CONNACK:
                return new ConnAckMessage(fixedHeader, (ConnAckVariableHeader) variableHeader);

            case SUBSCRIBE:
                return new SubscribeMessage(
                        fixedHeader,
                        (PackageIdVariableHeader) variableHeader,
                        (SubscribePayload) payload);

            case SUBACK:
                return new SubAckMessage(
                        fixedHeader,
                        (PackageIdVariableHeader) variableHeader,
                        (SubAckPayload) payload);

            case UNSUBSCRIBE:
                return new UnSubscribeMessage(
                        fixedHeader,
                        (PackageIdVariableHeader) variableHeader,
                        (UnSubscribePayload) payload);

            case PUBLISH:
                return new PublishMessage(
                        fixedHeader,
                        (PublishVariableHeader) variableHeader,
                        (ByteBuf) payload);

            case PUBACK:
            case UNSUBACK:
            case PUBREC:
            case PUBREL:
            case PUBCOMP:
                return new Message(fixedHeader, variableHeader);

            case PINGREQ:
            case PINGRESP:
            case DISCONNECT:
                return new Message(fixedHeader);

            default:
                throw new IllegalArgumentException("unknown message type: " + fixedHeader.getMessageType());
        }
    }

    private MQTTMesageFactory() { }
} 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 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 class ConnAckVariableHeader{
	private ConnectionStatus status;//返回给客户端的状态码
	private Boolean sessionPresent;//sessionPresent是告知客户端服务器是否存储了session的位
	
	public ConnAckVariableHeader(ConnectionStatus status, Boolean sessionPresent) {
		this.status = status;
		this.sessionPresent = sessionPresent;
	}
	public ConnectionStatus getStatus() {
		return status;
	}
	public void setStatus(ConnectionStatus status) {
		this.status = status;
	}
	public Boolean isSessionPresent() {
		return sessionPresent;
	}
	public void setSessionPresent(Boolean sessionPresent) {
		this.sessionPresent = sessionPresent;
	}
} 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 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 PackageIDManager {

	//包ID是两个字节，所以最大的是65535，最小是1
	private static final int MIN_MSG_ID = 1;		
	private static final int MAX_MSG_ID = 65535;
	private static int nextMsgId = MIN_MSG_ID - 1;
	private static Hashtable<Integer, Integer> inUseMsgIds = new Hashtable<Integer, Integer>();
		
	private int packgeID;//包ID	
	
	
	/**
	 * 获取包ID
	 * @return int
	 * @author zer0
	 * @version 1.0
	 * @date 2015-3-5
	 */
	public static synchronized int getNextMessageId(){
		int startingMessageId = nextMsgId;
		//循环两次是为了给异步出问题提供一个容错范围
		int loopCount = 0;
	    do {
	        nextMsgId++;
	        if ( nextMsgId > MAX_MSG_ID ) {
	            nextMsgId = MIN_MSG_ID;
	        }
	        if (nextMsgId == startingMessageId) {
	        	loopCount++;
	        	if (loopCount == 2) {
	        		throw new UnsupportedOperationException("获取不到可用的包ID");
	        	}
	        }
	    } while( inUseMsgIds.containsKey( new Integer(nextMsgId) ) );
	    Integer id = new Integer(nextMsgId);
	    inUseMsgIds.put(id, id);
	    return nextMsgId;
	}
	
	/**
	 * 释放不用的包ID
	 * @param msgId
	 * @author zer0
	 * @version 1.0
	 * @date 2015-3-3
	 */
	public synchronized static void releaseMessageId(int msgId) {
		inUseMsgIds.remove(new Integer(msgId));
	}
	
} 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 enum QoS {
	AT_MOST_ONCE  (0),
	AT_LEAST_ONCE (1),
	EXACTLY_ONCE  (2),
	RESERVE(3);
	
	final public int val;
	
	QoS(int val) {
		this.val = val;
	}
	
	/**
	 * 获取类型对应的值
	 * @return int
	 * @author zer0
	 * @version 1.0
	 * @date 2016-3-3
	 */
	public int value() {
		return val;
	}
	
	//通过读取到的整型来获取对应的QoS类型
	public static QoS valueOf(int i) {
		for(QoS q: QoS.values()) {
			if (q.val == i)
				return q;
		}
		throw new IllegalArgumentException("Qos值无效: " + i);
	}
} 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 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 IdentityAuthenticator implements IAuthenticator {

	@Override
	public boolean checkValid(String username, String password) {
		//该处连接数据库，到数据库查询是否有该用户，有则通过验证
		int ret=0;
		DruidPooledConnection conn=null;
		PreparedStatement statement = null;
		ResultSet resultSet=null;
		try {
			conn=DBConnection.getInstance().openConnection();
			String sqlString="select * from zer0_user where username=? and password=?";
			statement = conn.prepareStatement(sqlString);
			statement.setString(1, username);
			statement.setString(2, password);
			resultSet=statement.executeQuery();
			while (resultSet.next()) {
				ret=1;
				break;
			}
		} catch (Exception e) {
			e.printStackTrace();
		}finally{
			DBConnection.getInstance().closeConnection(conn, statement, resultSet);
		}
		if (ret == 1) {
			return true;
		}else {
			return false;
		}
	}

} public class MapDBPersistentStore implements IMessagesStore, ISessionStore {

	private final static Logger Log = Logger.getLogger(MapDBPersistentStore.class);
	
	//为Session保存的的可能需要重发的消息
	private ConcurrentMap<String, List<PublishEvent>> persistentOfflineMessage;
	//为Qos1和Qos2临时保存的消息
	private ConcurrentMap<String, PublishEvent> persistentQosTempMessage;
	//为Qos2临时保存的PubRel消息
	private ConcurrentMap<String, PubRelEvent> persistentPubRelTempMessage;
	//持久化存储session和与之对应的subscription Set
	private ConcurrentMap<String, Set<Subscription>> persistentSubscriptionStore;
	//持久化的Retain
	private ConcurrentMap<String, StoredMessage> retainedStore;
	//保存publish包ID
	private ConcurrentMap<String, Integer> publishPackgeIDStore;
	//保存pubRec包ID
	private ConcurrentMap<String, Integer> pubRecPackgeIDStore;
	
	private DB m_db;
	
	@Override
	public void initStore() {
		 String STORAGE_FILE_PATH =  System.getProperty("user.dir") + File.separator + MqttTool.getProperty("path");
		 Log.info("存储文件的初始化位置"+STORAGE_FILE_PATH);
	     File tmpFile;
	     try {
	    	 tmpFile = new File(STORAGE_FILE_PATH);
	         tmpFile.createNewFile();
	         m_db = DBMaker.newFileDB(tmpFile).make();
		     persistentOfflineMessage = m_db.getHashMap("offline");
		     persistentQosTempMessage = m_db.getHashMap("publishTemp");
		     persistentPubRelTempMessage = m_db.getHashMap("pubRelTemp");
		     persistentSubscriptionStore = m_db.getHashMap("subscriptions");
		     retainedStore = m_db.getHashMap("retained");
		     publishPackgeIDStore = m_db.getHashMap("publishPID");
		     pubRecPackgeIDStore = m_db.getHashMap("pubRecPID");
	     } catch (IOException ex) {
	         Log.error(null, ex);
	     }     
	 }
	
	@Override
	public boolean searchSubscriptions(String clientID) {
		return persistentSubscriptionStore.containsKey(clientID);
	}

	@Override
	public void wipeSubscriptions(String clientID) {
		persistentSubscriptionStore.remove(clientID);
		m_db.commit();
	}

	@Override
	public void addNewSubscription(Subscription newSubscription, String clientID) {
		Log.info("添加新订阅，订阅:" + newSubscription + ",客户端ID:" + clientID );
		 if (!persistentSubscriptionStore.containsKey(clientID)) {
	            Log.info("客户端ID{" + clientID + "}不存在订阅集 , 为它创建订阅集");
	            persistentSubscriptionStore.put(clientID, new HashSet<Subscription>());
	     }
		 
		 Set<Subscription> subs = persistentSubscriptionStore.get(clientID);
		 if (!subs.contains(newSubscription)) {
	            Log.info("更新客户端" + clientID + "的订阅集");
	            Subscription existingSubscription = null;
	            //遍历订阅集里所有的订阅，查看是否有相同topic的订阅，有的话，移除之前的订阅，添加新的
	            for (Subscription scanSub : subs) {
	                if (newSubscription.getTopicFilter().equals(scanSub.getTopicFilter())) {
	                    existingSubscription = scanSub;
	                    break;
	                }
	            }
	            if (existingSubscription != null) {
	                subs.remove(existingSubscription);
	            }
	            subs.add(newSubscription);
	            persistentSubscriptionStore.put(clientID, subs);
	            Log.debug("客户端" + clientID + "的订阅集现在是这样的" + subs);
	      }
		  m_db.commit();
	}

	@Override
	public void removeSubscription(String topic, String clientID) {
		Log.info("删除客户端" + clientID + "的" + topic + "订阅");
		if (!persistentSubscriptionStore.containsKey(clientID)) {
            Log.debug("没客户端ID" + clientID + " , 无法删除");
            return;
        }
		Set<Subscription> subs = persistentSubscriptionStore.get(clientID);
		Subscription existingSubscription = null;
		for (Subscription subscription : subs) {
			String topicfilter = subscription.getTopicFilter();
			if (topicfilter.equals(topic)) {
				existingSubscription = subscription;
			}
		}
		if (existingSubscription != null) {
            subs.remove(existingSubscription);
        }
		m_db.commit();
	}

	@Override
	public List<PublishEvent> listMessagesInSession(String clientID) {
		List<PublishEvent> allEvents = new ArrayList<PublishEvent>();
		List<PublishEvent> storeEvents = persistentOfflineMessage.get(clientID);
		//如果该client无离线消息，则把storeEvents设置为空集合
		if (storeEvents == null) {
			storeEvents = Collections.<PublishEvent>emptyList();
		}
		for (PublishEvent event : storeEvents) {
			allEvents.add(event);
		}
		return allEvents;
	}

	@Override
	public void removeMessageInSessionForPublish(String clientID,
			Integer packgeID) {
		List<PublishEvent> events = persistentOfflineMessage.get(clientID);
		if (events == null) {
			return;
		}
		PublishEvent toRemoveEvt = null;
		for (PublishEvent evt : events) {
	            if (evt.getPackgeID()== packgeID) {
	                toRemoveEvt = evt;
	            }
	     }
		events.remove(toRemoveEvt);
		persistentOfflineMessage.put(clientID, events);
		m_db.commit();
	}

	@Override
	public void storeMessageToSessionForPublish(PublishEvent pubEvent) {
		 	List<PublishEvent> storedEvents;
	        String clientID = pubEvent.getClientID();
	        if (!persistentOfflineMessage.containsKey(clientID)) {
	            storedEvents = new ArrayList<PublishEvent>();
	        } else {
	            storedEvents = persistentOfflineMessage.get(clientID);
	        }
	        storedEvents.add(pubEvent);
	        persistentOfflineMessage.put(clientID, storedEvents);
	        m_db.commit();
	}

	@Override
	public void storeQosPublishMessage(String publishKey, PublishEvent pubEvent) {
		persistentQosTempMessage.put(publishKey, pubEvent);
		m_db.commit();
	}

	@Override
	public void removeQosPublishMessage(String publishKey) {
		persistentQosTempMessage.remove(publishKey);
		m_db.commit();
	}
	
	@Override
	public PublishEvent searchQosPublishMessage(String publishKey) {
		return persistentQosTempMessage.get(publishKey);
	}
	
	@Override
	public void storePubRelMessage(String pubRelKey, PubRelEvent pubRelEvent) {
		persistentPubRelTempMessage.put(pubRelKey, pubRelEvent);
		m_db.commit();
	}

	@Override
	public void removePubRelMessage(String pubRelKey) {
		persistentPubRelTempMessage.remove(pubRelKey);
		m_db.commit();
	}

	@Override
	public PubRelEvent searchPubRelMessage(String pubRelKey) {
		return persistentPubRelTempMessage.get(pubRelKey);
	}

	@Override
	public void storeRetained(String topic, ByteBuf message, QoS qos) {
		//将ByteBuf转变为byte[]
		byte[] messageBytes = new byte[message.readableBytes()];
		message.getBytes(message.readerIndex(), messageBytes);
		if (messageBytes.length <= 0) {
			retainedStore.remove(topic);
		} else {
			StoredMessage storedMessage = new StoredMessage(messageBytes, qos, topic);
			retainedStore.put(topic, storedMessage);
		}
		m_db.commit();
	}

	@Override
	public void cleanRetained(String topic) {
		retainedStore.remove(topic);
		m_db.commit();
	}

	@Override
	public Collection<StoredMessage> searchRetained(String topic) {
		List<StoredMessage> results = new ArrayList<StoredMessage>();
		 for (Map.Entry<String, StoredMessage> entry : retainedStore.entrySet()) {
	            StoredMessage storedMsg = entry.getValue();
	            if (SubscribeStore.matchTopics(entry.getKey(), topic)) {
	                results.add(storedMsg);
	            }
	      }
		return results;
	}

	@Override
	public void storePublicPackgeID(String clientID, Integer packgeID) {
		publishPackgeIDStore.put(clientID, packgeID);
		m_db.commit();
	}

	@Override
	public void removePublicPackgeID(String clientID) {
		publishPackgeIDStore.remove(clientID);
		m_db.commit();
	}

	@Override
	public void storePubRecPackgeID(String clientID, Integer packgeID) {
		pubRecPackgeIDStore.put(clientID, packgeID);
		m_db.commit();
	}
	
	@Override
	public void removePubRecPackgeID(String clientID) {
		pubRecPackgeIDStore.remove(clientID);
		m_db.commit();
	}

} public interface IAuthenticator {

	/**
	 * 校验用户名和密码是否正确
	 * @param username
	 * @param password
	 * @return boolean
	 * @author zer0
	 * @version 1.0
	 * @date 2015-3-8
	 */
	boolean checkValid(String username, String password);
	
} public interface IMessagesStore {

	public static class StoredMessage implements Serializable {
		final QoS qos;
        final byte[] payload;
        final String topic;

        public StoredMessage(byte[] message, QoS qos, String topic) {
            this.qos = qos;
            this.payload = message;
            this.topic = topic;
        }

        public QoS getQos() {
            return qos;
        }

        public byte[] getPayload() {
			return payload;
		}

		public String getTopic() {
            return topic;
        }
		
    }
	
	/**
	 * 初始化存储
	 * @author zer0
	 * @version 1.0
	 * @date 2015-11-30
	 */
	void initStore();
	
	/**
	 * 返回某个clientID的离线消息列表
	 * @param clientID
	 * @author zer0
	 * @version 1.0
	 * @date 2015-05-18
	 */
	List<PublishEvent> listMessagesInSession(String clientID);
	

	/**
	 * 在重发以后，移除publish的离线消息事件
	 * @param clientID
	 * @param packgeID
	 * @author zer0
	 * @version 1.0
	 * @date 2015-05-18
	 */
	void removeMessageInSessionForPublish(String clientID, Integer packgeID);
	
	/**
	 * 存储publish的离线消息事件，为CleanSession=0的情况做重发准备
	 * @param pubEvent
	 * @author zer0
	 * @version 1.0
	 * @date 2015-05-21
	 */
	void storeMessageToSessionForPublish(PublishEvent pubEvent);

	/**
	 * 存储Publish的包ID
	 * @param clientID
	 * @param packgeID
	 * @author zer0
	 * @version 1.0
	 * @date 2015-05-21
	 */
	void storePublicPackgeID(String clientID, Integer packgeID);
	
	/**
	 * 移除Publish的包ID
	 * @param clientID
	 * @author zer0
	 * @version 1.0
	 * @date 2015-05-21
	 */
	void removePublicPackgeID(String clientID);
	
	/**
	 * 移除PubRec的包ID
	 * @param clientID
	 * @author zer0
	 * @version 1.0
	 * @date 2015-05-21
	 */
	void removePubRecPackgeID(String clientID);
	
	/**
	 * 存储PubRec的包ID
	 * @param clientID
	 * @param packgeID
	 * @author zer0
	 * @version 1.0
	 * @date 2015-05-21
	 */
	void storePubRecPackgeID(String clientID, Integer packgeID);
	
	/**
	 * 当Qos>0的时候，临时存储Publish消息，用于重发
	 * @param publishKey
	 * @param pubEvent
	 * @author zer0
	 * @version 1.0
	 * @date 2015-05-21
	 */
	void storeQosPublishMessage(String publishKey, PublishEvent pubEvent);
	
	/**
	 * 在收到对应的响应包后，删除Publish消息的临时存储
	 * @param publishKey
	 * @author zer0
	 * @version 1.0
	 * @date 2015-05-21
	 */
	void removeQosPublishMessage(String publishKey);

	/**
	 * 获取临时存储的Publish消息，在等待时间过后未收到对应的响应包，则重发该Publish消息
	 * @param publishKey
	 * @return PublishEvent
	 * @author zer0
	 * @version 1.0
	 * @date 2015-11-28
	 */
	PublishEvent searchQosPublishMessage(String publishKey);
	
	/**
	 * 当Qos=2的时候，临时存储PubRel消息，在未收到PubComp包时用于重发
	 * @param pubRelKey
	 * @param pubRelEvent
	 * @author zer0
	 * @version 1.0
	 * @date 2015-11-28
	 */
	void storePubRelMessage(String pubRelKey, PubRelEvent pubRelEvent);
	
	/**
	 * 在收到对应的响应包后，删除PubRel消息的临时存储
	 * @param pubRelKey
	 * @author zer0
	 * @version 1.0
	 * @date 2015-11-28
	 */
	void removePubRelMessage(String pubRelKey);

	/**
	 * 获取临时存储的PubRel消息，在等待时间过后未收到对应的响应包，则重发该PubRel消息
	 * @param pubRelKey
	 * @author zer0
	 * @version 1.0
	 * @date 2015-11-28
	 */
	PubRelEvent searchPubRelMessage(String pubRelKey);
	
	/**
	 * 持久化存储保留Retain为1的指定topic的最新信息，该信息会在新客户端订阅某主题的时候发送给此客户端
	 * @param topic
	 * @param message
	 * @param qos
	 * @author zer0
	 * @version 1.0
	 * @date 2015-05-26
	 */
    void storeRetained(String topic, ByteBuf message, QoS qos);
    
    /**
	 * 删除指定topic的Retain信息
	 * @param topic
	 * @author zer0
	 * @version 1.0
	 * @date 2015-05-26
	 */
    void cleanRetained(String topic);
    
    /**
	 * 从Retain中搜索对应topic中保存的信息
	 * @param topic
	 * @author zer0
	 * @version 1.0
	 * @date 2015-11-27
	 */
    Collection<StoredMessage> searchRetained(String topic);
} public interface ISessionStore {

	/**
	 * 查看是否已储存了该客户端ID，如果存储了则返回true
	 * @param clientID
	 * @return boolean
	 * @author zer0
	 * @version 1.0
	 * @date 2015-05-07
	 */
	boolean searchSubscriptions(String clientID);
	
	/**
	 * 清理某个ID所有订阅信息
	 * @param clientID
	 * @author zer0
	 * @version 1.0
	 * @date 2015-05-19
	 */
	void wipeSubscriptions(String clientID);
	

	/**
	 * 添加某个订阅消息到存储
	 * @param newSubscription
	 * @param clientID
	 * @author zer0
	 * @version 1.0
	 * @date 2015-05-25
	 */
	void addNewSubscription(Subscription newSubscription, String clientID);
	
	/**
	 * 从会话的持久化存储中移除某个订阅主题中的某个client
	 * @param topic
	 * @param clientID
	 * @author zer0
	 * @version 1.0
	 * @date 2015-05-25
	 */
	void removeSubscription(String topic, String clientID);
	
} public class PubRelEvent {
	String clientID;
	int packgeID;
	
	public PubRelEvent(String clientID, Integer pkgID){
		this.clientID = clientID;
		this.packgeID = pkgID;
	}

	public String getClientID() {
		return clientID;
	}

	public void setClientID(String clientID) {
		this.clientID = clientID;
	}

	public int getPackgeID() {
		return packgeID;
	}

	public void setPackgeID(int packgeID) {
		this.packgeID = packgeID;
	}
	
	
	
} public class PublishEvent implements Serializable{
	String topic;
	QoS qos;
	byte[] message;
	boolean retain;
	String clientID;
	//针对Qos1和Qos2
	int packgeID;
	
	public PublishEvent(String topic, QoS qos, byte[] message, boolean retain, String clientID, Integer pkgID){
		this.topic = topic;
		this.qos = qos;
		this.message = message;
		this.retain = retain;
		this.clientID = clientID;
		if (qos != QoS.AT_MOST_ONCE) {
			this.packgeID = pkgID;
		}
	}

	public String getTopic() {
		return topic;
	}

	public void setTopic(String topic) {
		this.topic = topic;
	}

	public QoS getQos() {
		return qos;
	}

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

	public byte[] getMessage() {
		return message;
	}

	public void setMessage(byte[] message) {
		this.message = message;
	}

	public boolean isRetain() {
		return retain;
	}

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

	public String getClientID() {
		return clientID;
	}

	public void setClientID(String clientID) {
		this.clientID = clientID;
	}

	public int getPackgeID() {
		return packgeID;
	}

	public void setPackgeID(int packgeID) {
		this.packgeID = packgeID;
	}

} public class RePubRelJob implements Job{
	
	@Override
	public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException {
		//取出参数，参数为ProtocolProcess，调用此类的函数
		JobDataMap dataMap = jobExecutionContext.getJobDetail().getJobDataMap();
		ProtocolProcess process = (ProtocolProcess) dataMap.get("ProtocolProcess");
		String pubRelKey = (String) dataMap.get("pubRelKey");
		process.reUnKnowPubRelMessage(pubRelKey);
	}
	
} public class RePublishJob implements Job{
	
	private final static Logger Log = Logger.getLogger(RePublishJob.class);
	int count = 0;
	
	@Override
	public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException {
		//取出参数，参数为ProtocolProcess，调用此类的函数
		JobDataMap dataMap = jobExecutionContext.getJobDetail().getJobDataMap();
		ProtocolProcess process = (ProtocolProcess) dataMap.get("ProtocolProcess");
		String publishKey = (String) dataMap.get("publishKey");
		process.reUnKnowPublishMessage(publishKey);
	}
	
} public class SubscribeStore {
    
	private TreeNode root = new TreeNode(null);
	
	 /**
	  * 添加新的订阅到订阅树里
	  * @param newSubscription
	  * @author zer0
	  * @version 1.0
      * @date 2015-04-19
	  */
	public void addSubscrpition(Subscription newSubscription){	
		List<TreeNode> treeNodes = searchNodeList(newSubscription.topicFilter);
		for (TreeNode t : treeNodes) {
			t.addSubscription(newSubscription);
		}
	} 

	 /**
	  * 根据topic搜索节点，若无此节点则创建,最后返回一个所搜索的所有节点的列表
	  * @param topic
	  * @return List<TreeNode>
	  * @author zer0
	  * @version 1.0
      * @date 2015-04-25
	  */
	public List<TreeNode> searchNodeList(String topic){
		List<Token> tokens = new ArrayList<Token>();
		List<TreeNode> childList = new ArrayList<TreeNode>();
		
		try {
			tokens = parseTopic(topic);
		} catch (ParseException e) {
			e.printStackTrace();
		}
		TreeNode current = root;
		for (int i = 0; i < tokens.size(); i++) {
			Token token = tokens.get(i);
			TreeNode matchingChildren = current;
 
			//TODO 为实现简便，此处违反协议
			if (token == Token.SINGLE) {
				//判断‘+’是不是在队列最后一个
				if (i == tokens.size() - 1) {
					childList.addAll(matchingChildren.children);
				}
				return childList;
			}
			
			//遇到#，证明已搜索到底部，不再生成节点
			if (token == Token.MULTI) {	
				if (i == tokens.size() - 1) {
					//判断‘#’是不是在队列最后一个
					childList.add(matchingChildren);
					childList.addAll(matchingChildren.getAllDescendant());
				}
				return childList;	
			}			
			
			if ((matchingChildren = current.childWithToken(token)) != null) {
				current = matchingChildren;
			} else {
				matchingChildren = new TreeNode(current);
				matchingChildren.setToken(token);
				current.addChild(matchingChildren);
				current = matchingChildren;
			}
			
			//如果该token是最后一个，则添加到返回列表里
			if (i == tokens.size() - 1) {
				childList.add(current);
			}
		}
		return childList;
	}
	
	 /**
	  * 解析topic,从topic获取到对应的客户端ID群
	  * @param topic
	  * @return List<Subscription>
	  * @author zer0
	  * @version 1.0
     * @date 2015-05-04
	  */
	public List<Subscription> getClientListFromTopic(String topic) {
		List<Token> tokens;
		try {
			tokens = parseTopic(topic);
		} catch (ParseException e) {
			return Collections.emptyList();
		}
		Queue<Token> tokenQueue = new LinkedBlockingDeque<Token>(tokens);
		List<Subscription> matchingSubs = new ArrayList<Subscription>();
		root.getSubscription(tokenQueue, matchingSubs);
		return matchingSubs;
	}
	
	 /**
	  * 把某个clientID从订阅树里移走
	  * @param clientID
	  * @author zer0
	  * @version 1.0
      * @date 2015-05-04
	  */
	public void removeForClient(String clientID) {
	    root.removeClientSubscription(clientID);
	}

	 /**
	  * 从订阅结构树中移除某个订阅主题中的某个client
	  * @param topic
	  * @param clientID
	  * @author zer0
	  * @version 1.0
      * @date 2015-06-29
	  */
	public void removeSubscription(String topic, String clientID){
		List<TreeNode> treeNodes = searchNodeList(topic);
		for (TreeNode t : treeNodes) {
			t.removeClientSubscription(clientID);
			//如果某个节点的订阅量为空，并且该节点无子节点，则删除它
			if (t.subscriptions.size() == 0 && t.children.size() == 0) {
				t.parent.children.remove(t);
			}
		}
	}
	
	/**
	 * 比较两个Topic字段是否相等，需处理#，+等情况
	 * @param retainTopic
	 * @param subscriptionTopic
	 * @return boolean
	 * @author zer0
	 * @version 1.0
	 * @date 2015-12-1
	 */
	 public static boolean matchTopics(String retainTopic, String subscriptionTopic) {
		 try {
	            List<Token> retainTokens = SubscribeStore.parseTopic(retainTopic);
	            List<Token> subscriptionTokens = SubscribeStore.parseTopic(subscriptionTopic);
	            int i = 0;
	            Token subToken = null;
	            for (; i< subscriptionTokens.size(); i++) {
	                subToken = subscriptionTokens.get(i);
	                if (subToken != Token.MULTI && subToken != Token.SINGLE) {
	                    if (i >= retainTokens.size()) {
	                        return false;
	                    }
	                    Token msgToken = retainTokens.get(i);
	                    if (!msgToken.equals(subToken)) {
	                        return false;
	                    }
	                } else {
	                    if (subToken == Token.MULTI) {
	                        return true;
	                    }
	                    if (subToken == Token.SINGLE) {
	                        //跳过
	                    }
	                }
	            }
	            return i == retainTokens.size();
	        } catch (ParseException ex) {
	            ex.printStackTrace();
	            throw new RuntimeException(ex);
	        }
	 }

	 /**
	  *  解析topic，根据协议得到对应的token列表
	  * @param topic
	  * @return List<Token>
	  * @author zer0
	  * @version 1.0
      * @date 2015-06-29
	  */
	static List<Token> parseTopic(String topic) throws ParseException{
		List<Token> tokens = new ArrayList<Token>();
		String[] token = topic.split("/");
		
		if (token.length == 0) {
			 tokens.add(Token.EMPTY);
		}
		
		for (int i = 0; i < token.length; i++) {
			String s = token[i];
			if (s.isEmpty()) {
				tokens.add(Token.EMPTY);
			}else if (s.equals("#")) {
				if (i != token.length - 1) {
					throw new ParseException("无效的的topic，" + s + "#必须放在分隔符的最尾", i);
				}else {
					tokens.add(Token.MULTI);
				}	
			}else if (s.contains("#")){
				throw new ParseException("无效的topic"+s, i);
			}else if (s.equals("+")) {
				//违反协议，不好实现，视情况看以后是否处理
				if (i != token.length - 1) {
					throw new ParseException("无效的的topic，" + s + "此服务器暂不支持通配符‘+’作为中层分隔符", i);
				}else {
					tokens.add(Token.SINGLE);
				}
			}else if (s.contains("+")) {
				throw new ParseException("无效的topic"+s, i);
			}else{
				tokens.add(new Token(s));
			}
		}
		return tokens;
	} 
} public class Subscription implements Serializable{

	private static final long serialVersionUID = 6159635655653102856L;
	QoS requestedQos; //max QoS acceptable
    String topicFilter;
    String clientID;
    boolean cleanSession;
    boolean active = true;
    
    public Subscription(String clientID, String topicFilter, QoS requestedQos, boolean cleanSession) {
    	this.clientID = clientID;
        this.requestedQos = requestedQos;
        this.topicFilter = topicFilter;
        this.cleanSession = cleanSession;
    }
    
    public QoS getRequestedQos() {
		return requestedQos;
	}

	public void setRequestedQos(QoS requestedQos) {
		this.requestedQos = requestedQos;
	}

	public String getTopicFilter() {
		return topicFilter;
	}

	public void setTopicFilter(String topicFilter) {
		this.topicFilter = topicFilter;
	}
 
	public boolean isCleanSession() {
		return cleanSession;
	}

	public void setCleanSession(boolean cleanSession) {
		this.cleanSession = cleanSession;
	}

	public boolean isActive() {
		return active;
	}

	public void setActive(boolean active) {
		this.active = active;
	}

	public void setClientID(String clientID) {
		this.clientID = clientID;
	}

	public String getClientID() {
		return clientID;
	}
 
    @Override
    public String toString() {
        return String.format("[filter:%s, cliID: %s, qos: %s, active: %s]", this.topicFilter, this.clientID, this.requestedQos, this.active);
    }
} public class Constant {
    
    
    public static final String PING_ARRIVE = "pingIsArrive";//心跳包到达
    public static final String CONNECT_ARRIVE = "connectIsArrive";//连接数据包到达
} public class QuartzManager {

	private final static Logger Log = Logger.getLogger(QuartzManager.class);
	
	private static SchedulerFactory sf = new StdSchedulerFactory();
	
	/**
	 * 添加调度任务
	 * @param jobName
	 * @param jobGroupName
	 * @param triggerName
	 * @param triggerGroupName
	 * @param jobClass
	 * @param time
	 * @param jobParam
	 * @param count重复次数
	 * @author zer0
	 * @version 1.0
	 * @date 2015-11-28
	 */
	@SuppressWarnings("unchecked")
	public static void addJob(String jobName, String jobGroupName,  
            String triggerName, String triggerGroupName, Class jobClass,  
            int time, int count, Map<String, Object> jobParam) {  
        try {  
            Scheduler sched = sf.getScheduler();  
            JobDetail job = newJob(jobClass).withIdentity(jobName, jobGroupName).build();// 任务名，任务组，任务执行类  
            if (jobParam != null && jobParam.size() > 0) {
				//给job添加参数
            	job.getJobDataMap().putAll(jobParam);
			}
            // 触发器  
            Trigger trigger = TriggerBuilder.newTrigger().withIdentity(triggerName, triggerGroupName)
    	            .startAt(futureDate(time, IntervalUnit.SECOND))
    	            .withSchedule(SimpleScheduleBuilder
    	            		.simpleSchedule()
    	            		.withIntervalInSeconds(time)
    	            		.withRepeatCount(count))
    	            .build();
            Date ft = sched.scheduleJob(job, trigger);
            Log.info(jobName + "启动于" + ft);
            // 启动  
            if (!sched.isShutdown()) {  
                sched.start();  
            }  
        } catch (Exception e) {  
            throw new RuntimeException(e);  
        }  
    }
	
	/**
	 * 删除调度任务
	 * @param jobName
	 * @param jobGroupName
	 * @param triggerName
	 * @param triggerGroupName
	 * @author zer0
	 * @version 1.0
	 * @date 2015-11-28
	 */
	public static void removeJob(String jobName,String jobGroupName,  
            String triggerName,String triggerGroupName){
		try {
			Scheduler sched = sf.getScheduler();
			TriggerKey triggerKey = new TriggerKey(triggerName, triggerGroupName);
			JobKey jobKey = new JobKey(jobName, jobGroupName);
			sched.pauseTrigger(triggerKey);//停止触发器  
			sched.unscheduleJob(triggerKey);//移除触发器  
			sched.deleteJob(jobKey);//删除任务
		} catch (SchedulerException e) {
			e.printStackTrace();
		}
	}
	
	/**
	 * 重置调度任务
	 * @param jobName
	 * @param jobGroupName
	 * @param triggerName
	 * @param triggerGroupName
	 * @author zer0
	 * @version 1.0
	 * @date 2015-12-3
	 */
	public static void resetJob(String jobName,String jobGroupName,  
            String triggerName,String triggerGroupName, Class jobClass,  
            int time, int count, Map<String, Object> jobParam){
		removeJob(jobName, jobGroupName, triggerName, triggerGroupName);
		addJob(jobName, jobGroupName, triggerName, triggerGroupName, jobClass, time, count, jobParam);
	}
 
	
} public class StringTool {

	/**
	 * 判断字符串是否为空
	 * 
	 * @param str
	 * @return boolean
	 * @author zer0
	 * @version 1.0
	 * @date 2015-2-19
	 */
	public static boolean isBlank(String str){
		if(str==null){
			return true;
		}
		if(str.trim().length()<1){
			return true;
		}
		if(str.trim().equals("")){
			return true;
		}
		if(str.trim().toLowerCase().equals("null")){
			return true;
		}
		return false;
	}
	
	/**
	 * 将字符串转换为byte数组
	 * 
	 * @param string
	 * @return byte[]
	 * @author zer0
	 * @version 1.0
	 * @date 2015-3-3
	 */
	public static byte[] stringToByte(String string) {
		if (string == null) {
			return new byte[0];
		}
		ByteArrayOutputStream byteOut = new ByteArrayOutputStream();
		DataOutputStream dos = new DataOutputStream(byteOut);
		try {
			dos.writeUTF(string);
		} catch (IOException e) {
			return new byte[0];
		}
		return byteOut.toByteArray();
	}
	
	/**
	 * 随机生成字符串
	 * 
	 * @param length
	 * @return String
	 * @author zer0
	 * @version 1.0
	 * @date 2015-3-7
	 */
	 public static String generalRandomString(int length){
	     String str="abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789";
	     Random random=new Random();
	     StringBuffer sb=new StringBuffer();
	     for(int i=0;i<length;i++){
	       int number=random.nextInt(62);
	       sb.append(str.charAt(number));
	     }
	     return sb.toString();
	 }
	 
	 /**
		 * 随机生成一个mac地址
		 * 
		 * @return String
		 * @author zer0
		 * @version 1.0
		 * @date  2015-7-7
		 */
	 public static String generalMacString(){
		 String str = "ABCDEF0123456789";
		 Random random = new Random();
		 StringBuffer sb = new StringBuffer();
		 int macGroupSize = 6;//mac地址长度为6组16进制
		 int macOneGroup = 2;//一组有两个字符
		 for (int i = 0; i < macGroupSize; i++) {
			 for (int j = 0; j < macOneGroup; j++) {
					int number = random.nextInt(16);
					sb.append(str.charAt(number));
			}
		    sb.append("-");
		}
		String macString = sb.toString().substring(0, sb.length()-1);
		return macString;
	 }
	 
	 /**
		 * 判断字符串是否为mac地址
		 * 
		 * @param str
		 * @return boolean
		 * @author zer0
		 * @version 1.0
		 * @date  2015-7-7
		 */
	 public static boolean isMacString(String str){
		 String pattern = "^([A-Fa-f\\d]{2}[-:])([A-Fa-f\\d]{2}[-:]){4}([A-Fa-f\\d]{2})$";
		 Pattern pat = Pattern.compile(pattern); 
		 Matcher mat = pat.matcher(str);
		 Boolean isFind = mat.find();
		 if (isFind) {
			return true;
		}else {
			return false;
		}
	 }
	
	 public static void main(String[] args){
		System.out.println(StringTool.generalRandomString(32));
	 } 
}
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.handler.timeout.IdleStateHandler;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import org.apache.log4j.Logger;
import com.syxy.protocol.mqttImp.MQTTMesageFactory;
import com.syxy.protocol.mqttImp.message.ConnAckMessage;
import com.syxy.protocol.mqttImp.message.ConnAckMessage.ConnectionStatus;
import com.syxy.protocol.mqttImp.message.ConnAckVariableHeader;
import com.syxy.protocol.mqttImp.message.ConnectMessage;
import com.syxy.protocol.mqttImp.message.FixedHeader;
import com.syxy.protocol.mqttImp.message.Message;
import com.syxy.protocol.mqttImp.message.PackageIDManager;
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.QoS;
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.TopicSubscribe;
import com.syxy.protocol.mqttImp.message.UnSubscribeMessage;
import com.syxy.protocol.mqttImp.process.Impl.IdentityAuthenticator;
import com.syxy.protocol.mqttImp.process.Impl.dataHandler.MapDBPersistentStore;
import com.syxy.protocol.mqttImp.process.Interface.IAuthenticator;
import com.syxy.protocol.mqttImp.process.Interface.IMessagesStore;
import com.syxy.protocol.mqttImp.process.Interface.ISessionStore;
import com.syxy.protocol.mqttImp.process.event.PubRelEvent;
import com.syxy.protocol.mqttImp.process.event.PublishEvent;
import com.syxy.protocol.mqttImp.process.event.job.RePubRelJob;
import com.syxy.protocol.mqttImp.process.event.job.RePublishJob;
import com.syxy.protocol.mqttImp.process.subscribe.SubscribeStore;
import com.syxy.protocol.mqttImp.process.subscribe.Subscription;
import com.syxy.util.Constant;
import com.syxy.util.QuartzManager;
import com.syxy.util.StringTool;
package com.syxy.protocol.mqttImp.process;





/**
 *  协议所有的业务处理都在此类，注释中所指协议为MQTT3.3.1协议英文版
 * 
 * @author zer0
 * @version 1.0
 * @date 2015-2-16
 */
public class ProtocolProcess {

	//遗嘱信息类
	static final class WillMessage {
        private final String topic;
        private final ByteBuf payload;
        private final boolean retained;
        private final QoS qos;

        public WillMessage(String topic, ByteBuf payload, boolean retained, QoS qos) {
            this.topic = topic;
            this.payload = payload;
            this.retained = retained;
            this.qos = qos;
        }

        public String getTopic() {
            return topic;
        }

        public ByteBuf getPayload() {
            return payload;
        }

        public boolean isRetained() {
            return retained;
        }

        public QoS getQos() {
            return qos;
        }
    }
	
	private final static Logger Log = Logger.getLogger(ProtocolProcess.class);
	
	private ConcurrentHashMap<Object, ConnectionDescriptor> clients = new ConcurrentHashMap<Object, ConnectionDescriptor>();// 客户端链接映射表
    //存储遗嘱信息，通过ID映射遗嘱信息
	private ConcurrentHashMap<String, WillMessage> willStore = new ConcurrentHashMap<>();
	
	private IAuthenticator authenticator;
	private IMessagesStore messagesStore;
	private ISessionStore sessionStore;
	private SubscribeStore subscribeStore;
	
	public ProtocolProcess(){
		MapDBPersistentStore storge = new MapDBPersistentStore();
		this.authenticator = new IdentityAuthenticator();