// Path: src/main/java/com/ystruct.middleware/stormmq/broker/AckManager.java
// public class AckManager {
//     //ConcurrentLinkedQueue是一个非阻塞的线程安全的队列,注意不要使用ConcurrentLinkQueue的size()方法,此方法会遍历所有的元素
//     private static ConcurrentLinkedQueue<SendResult> ackQueue = new ConcurrentLinkedQueue<SendResult>();
//     private static ConcurrentHashMap<String/*requestId*/,Channel> producerMap = new ConcurrentHashMap<String, Channel>();
// 
//     public static void pushRequest(String requestId,Channel channel){
//         producerMap.put(requestId,channel);
//     }
//     //查找到这个message发送来的channel,并从这里面删除它
//     public static Channel findChannel(String requestId){
//         Channel channel = producerMap.remove(requestId);
//         return  channel;
//     }
//     public static boolean pushAck(SendResult ack){
//        // System.out.println("pushAck");
//         return ackQueue.offer(ack);
//     }
//     public static  boolean pushAck(List<SendResult> acks){
//         boolean flag = false;
//         for(SendResult ack:acks){
//             flag = ackQueue.offer(ack);
//         }
//         return  flag;
//     }
//     public static SendResult getAck(){
//         return ackQueue.poll();
//     }
// }
// 
// Path: src/main/java/com/ystruct.middleware/stormmq/broker/SemaphoreManager.java
// public class SemaphoreManager {
// //每一个队列一个信号量管理里面的消费同步
//     private static Map<String,Semaphore> semMap = new HashMap<String, Semaphore>();
// 
//     //创建一个信号量
//     public static void createSemaphore(String key){
//         if(semMap.containsKey(key))
//             return;
//         Semaphore semaphore = new Semaphore(0);
//         semMap.put(key,semaphore);
//     }
//     //信号量的值加1
//     public static void increase(String key){
//         semMap.get(key).release();
//     }
//     //信号量的值减1
//     public static void descrease(String key){
//         try {
//             semMap.get(key).acquire();
//         }catch (InterruptedException e){
//             e.printStackTrace();
//         }
//     }
// }
// 
// Path: src/main/java/com/ystruct.middleware/stormmq/model/RequestResponseFromType.java
// public enum RequestResponseFromType {
//     Consumer,   //这个消息来自消费者.
//     Broker, //这个消息broker.
//     Produce  //此消息来自生产者.
// }
// 
// Path: src/main/java/com/ystruct.middleware/stormmq/model/ResponseType.java
// public enum ResponseType {
//     SendResult, //broker回应给生产者的ACK
//     Message, //消息 由broker发送给消费者的消息.
//     AckSubscript  //消费者订阅主题后,broker
// }
// 
// Path: src/main/java/com/ystruct.middleware/stormmq/model/StormResponse.java
// public class StormResponse implements Serializable{
//     private String requestId; //对应的回应的是哪个请求
//     private Object response;// 回应的消息
//     private RequestResponseFromType fromtype; //消息来自哪里
//     private ResponseType responseType; //响应的类型
// 
//     public String getRequestId() {
//         return requestId;
//     }
// 
//     public void setRequestId(String requestId) {
//         this.requestId = requestId;
//     }
// 
//     public Object getResponse() {
//         return response;
//     }
// 
//     public void setResponse(Object response) {
//         this.response = response;
//     }
// 
//     public RequestResponseFromType getFromtype() {
//         return fromtype;
//     }
// 
//     public void setFromtype(RequestResponseFromType fromtype) {
//         this.fromtype = fromtype;
//     }
// 
//     public ResponseType getResponseType() {
//         return responseType;
//     }
// 
//     public void setResponseType(ResponseType responseType) {
//         this.responseType = responseType;
//     }
// }
// 
// Path: src/main/java/com/ystruct.middleware/stormmq/smq/SendResult.java
// public class SendResult {
//     private String info;
//     private SendStatus status;
//     private String msgId;
// 
//     public String getInfo() {
//         return info;
//     }
// 
//     public void setInfo(String info) {
//         this.info = info;
//     }
// 
//     public SendStatus getStatus() {
//         return status;
//     }
// 
//     public void setStatus(SendStatus status) {
//         this.status = status;
//     }
// 
//     public String getMsgId() {
//         return msgId;
//     }
// 
//     public void setMsgId(String msgId) {
//         this.msgId = msgId;
//     }
//     public String toString(){
//         return new StringBuilder().append("msg ").append(this.msgId).append("  send ").append(this.status == SendStatus.SUCCESS ? "success":"fail").append("   info:").append(this.info).toString();
//     }
// 
// 
// }


import broker.AckManager;
import broker.SemaphoreManager;
import io.netty.channel.Channel;
import model.RequestResponseFromType;
import model.ResponseType;
import model.StormResponse;
import smq.SendResult;
package broker.netty;


/**
 * Created by yang on 16-12-1.
 */
public class AckSendThread implements Runnable {
    @Override
    public void run() {
// Path: src/main/java/com/ystruct.middleware/stormmq/broker/AckManager.java
// public class AckManager {
//     //ConcurrentLinkedQueue是一个非阻塞的线程安全的队列,注意不要使用ConcurrentLinkQueue的size()方法,此方法会遍历所有的元素
//     private static ConcurrentLinkedQueue<SendResult> ackQueue = new ConcurrentLinkedQueue<SendResult>();
//     private static ConcurrentHashMap<String/*requestId*/,Channel> producerMap = new ConcurrentHashMap<String, Channel>();
// 
//     public static void pushRequest(String requestId,Channel channel){
//         producerMap.put(requestId,channel);
//     }
//     //查找到这个message发送来的channel,并从这里面删除它
//     public static Channel findChannel(String requestId){
//         Channel channel = producerMap.remove(requestId);
//         return  channel;
//     }
//     public static boolean pushAck(SendResult ack){
//        // System.out.println("pushAck");
//         return ackQueue.offer(ack);
//     }
//     public static  boolean pushAck(List<SendResult> acks){
//         boolean flag = false;
//         for(SendResult ack:acks){
//             flag = ackQueue.offer(ack);
//         }
//         return  flag;
//     }
//     public static SendResult getAck(){
//         return ackQueue.poll();
//     }
// }
// 
// Path: src/main/java/com/ystruct.middleware/stormmq/broker/SemaphoreManager.java
// public class SemaphoreManager {
// //每一个队列一个信号量管理里面的消费同步
//     private static Map<String,Semaphore> semMap = new HashMap<String, Semaphore>();
// 
//     //创建一个信号量
//     public static void createSemaphore(String key){
//         if(semMap.containsKey(key))
//             return;
//         Semaphore semaphore = new Semaphore(0);
//         semMap.put(key,semaphore);
//     }
//     //信号量的值加1
//     public static void increase(String key){
//         semMap.get(key).release();
//     }
//     //信号量的值减1
//     public static void descrease(String key){
//         try {
//             semMap.get(key).acquire();
//         }catch (InterruptedException e){
//             e.printStackTrace();
//         }
//     }
// }
// 
// Path: src/main/java/com/ystruct.middleware/stormmq/model/RequestResponseFromType.java
// public enum RequestResponseFromType {
//     Consumer,   //这个消息来自消费者.
//     Broker, //这个消息broker.
//     Produce  //此消息来自生产者.
// }
// 
// Path: src/main/java/com/ystruct.middleware/stormmq/model/ResponseType.java
// public enum ResponseType {
//     SendResult, //broker回应给生产者的ACK
//     Message, //消息 由broker发送给消费者的消息.
//     AckSubscript  //消费者订阅主题后,broker
// }
// 
// Path: src/main/java/com/ystruct.middleware/stormmq/model/StormResponse.java
// public class StormResponse implements Serializable{
//     private String requestId; //对应的回应的是哪个请求
//     private Object response;// 回应的消息
//     private RequestResponseFromType fromtype; //消息来自哪里
//     private ResponseType responseType; //响应的类型
// 
//     public String getRequestId() {
//         return requestId;
//     }
// 
//     public void setRequestId(String requestId) {
//         this.requestId = requestId;
//     }
// 
//     public Object getResponse() {
//         return response;
//     }
// 
//     public void setResponse(Object response) {
//         this.response = response;
//     }
// 
//     public RequestResponseFromType getFromtype() {
//         return fromtype;
//     }
// 
//     public void setFromtype(RequestResponseFromType fromtype) {
//         this.fromtype = fromtype;
//     }
// 
//     public ResponseType getResponseType() {
//         return responseType;
//     }
// 
//     public void setResponseType(ResponseType responseType) {
//         this.responseType = responseType;
//     }
// }
// 
// Path: src/main/java/com/ystruct.middleware/stormmq/smq/SendResult.java
// public class SendResult {
//     private String info;
//     private SendStatus status;
//     private String msgId;
// 
//     public String getInfo() {
//         return info;
//     }
// 
//     public void setInfo(String info) {
//         this.info = info;
//     }
// 
//     public SendStatus getStatus() {
//         return status;
//     }
// 
//     public void setStatus(SendStatus status) {
//         this.status = status;
//     }
// 
//     public String getMsgId() {
//         return msgId;
//     }
// 
//     public void setMsgId(String msgId) {
//         this.msgId = msgId;
//     }
//     public String toString(){
//         return new StringBuilder().append("msg ").append(this.msgId).append("  send ").append(this.status == SendStatus.SUCCESS ? "success":"fail").append("   info:").append(this.info).toString();
//     }
// 
// 
// }


// Path: src/main/java/com/ystruct.middleware/stormmq/broker/netty/AckSendThread.java
import broker.AckManager;
import broker.SemaphoreManager;
import io.netty.channel.Channel;
import model.RequestResponseFromType;
import model.ResponseType;
import model.StormResponse;
import smq.SendResult;

package broker.netty;


/**
 * Created by yang on 16-12-1.
 */
public class AckSendThread implements Runnable {
    @Override
    public void run() {
        while (true){