// Path: src/main/java/com/neverwinterdp/sparkngin/Ack.java
// public class Ack implements Serializable {
//   static public enum Status { OK, ERROR, NOT_AVAIBLE }
// 
//   private Object messageId ;
//   private Status status;
//   private String message;
// 
//   public Object getMessageId() { return messageId; }
//   public void setMessageId(Object objectId) {
//     this.messageId = objectId;
//   }
// 
//   public Status getStatus() {
//     return status;
//   }
// 
//   public void setStatus(Status status) {
//     this.status = status;
//   }
// 
//   public String getMessage() {
//     return message;
//   }
// 
//   public void setMessage(String message) {
//     this.message = message;
//   }
// }


import io.netty.handler.codec.http.HttpContent;
import io.netty.handler.codec.http.HttpResponse;
import io.netty.handler.codec.http.QueryStringEncoder;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;
import com.neverwinterdp.message.Message;
import com.neverwinterdp.netty.http.client.AsyncHttpClient;
import com.neverwinterdp.netty.http.client.ResponseHandler;
import com.neverwinterdp.sparkngin.Ack;
      if(waitingAckMessages.size() >= bufferSize) {
        waitingAckMessages.wait(timeout);
        if(waitingAckMessages.size() >= bufferSize) {
          throw new TimeoutException("fail to send the message in " + timeout + "ms") ;
        }
      }
      QueryStringEncoder encoder = new QueryStringEncoder(path);
      encoder.addParam("data", toStringData(message));
      client.get(encoder.toString());
      sendCount++ ;
      String messageId = message.getHeader().getKey() ;
      waitingAckMessages.put(messageId, message) ;
    }
  }
  
  public void sendPost(Message message, long timeout) throws Exception {
    synchronized(waitingAckMessages) {
      if(waitingAckMessages.size() >= bufferSize) {
        waitingAckMessages.wait(timeout);
        if(waitingAckMessages.size() >= bufferSize) {
          throw new TimeoutException("fail to send the message in " + timeout + "ms") ;
        }
      }
      client.post(path, toBinData(message));
      sendCount++ ;
      String messageId = message.getHeader().getKey() ;
      waitingAckMessages.put(messageId, message) ;
    }
  }
  
// Path: src/main/java/com/neverwinterdp/sparkngin/Ack.java
// public class Ack implements Serializable {
//   static public enum Status { OK, ERROR, NOT_AVAIBLE }
// 
//   private Object messageId ;
//   private Status status;
//   private String message;
// 
//   public Object getMessageId() { return messageId; }
//   public void setMessageId(Object objectId) {
//     this.messageId = objectId;
//   }
// 
//   public Status getStatus() {
//     return status;
//   }
// 
//   public void setStatus(Status status) {
//     this.status = status;
//   }
// 
//   public String getMessage() {
//     return message;
//   }
// 
//   public void setMessage(String message) {
//     this.message = message;
//   }
// }


// Path: src/main/java/com/neverwinterdp/sparkngin/http/AbstractHttpSparknginClient.java
import io.netty.handler.codec.http.HttpContent;
import io.netty.handler.codec.http.HttpResponse;
import io.netty.handler.codec.http.QueryStringEncoder;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;
import com.neverwinterdp.message.Message;
import com.neverwinterdp.netty.http.client.AsyncHttpClient;
import com.neverwinterdp.netty.http.client.ResponseHandler;
import com.neverwinterdp.sparkngin.Ack;

      if(waitingAckMessages.size() >= bufferSize) {
        waitingAckMessages.wait(timeout);
        if(waitingAckMessages.size() >= bufferSize) {
          throw new TimeoutException("fail to send the message in " + timeout + "ms") ;
        }
      }
      QueryStringEncoder encoder = new QueryStringEncoder(path);
      encoder.addParam("data", toStringData(message));
      client.get(encoder.toString());
      sendCount++ ;
      String messageId = message.getHeader().getKey() ;
      waitingAckMessages.put(messageId, message) ;
    }
  }
  
  public void sendPost(Message message, long timeout) throws Exception {
    synchronized(waitingAckMessages) {
      if(waitingAckMessages.size() >= bufferSize) {
        waitingAckMessages.wait(timeout);
        if(waitingAckMessages.size() >= bufferSize) {
          throw new TimeoutException("fail to send the message in " + timeout + "ms") ;
        }
      }
      client.post(path, toBinData(message));
      sendCount++ ;
      String messageId = message.getHeader().getKey() ;
      waitingAckMessages.put(messageId, message) ;
    }
  }
  
  public void onFailedMessage(Ack ack, Message message) {