// Path: src/main/java/com/ystruct.middleware/stormmq/model/InvokeFuture.java
// public class InvokeFuture<T> {
//     private Semaphore semaphore = new Semaphore(0);
//     private T result;
//     private List<InvokeListener<T>> listeners = new ArrayList<InvokeListener<T>>();
//     private String requestId;
//     private Throwable cause;
//     public void setCause(Throwable cause){
//         this.cause = cause;
//         notifyListeners();
//         semaphore.release(Integer.MAX_VALUE - semaphore.availablePermits());
// 
//     }
//     public String getRequestId() {
//         return requestId;
//     }
// 
//     public void setRequestId(String requestId) {
//         this.requestId = requestId;
//     }
//     public InvokeFuture() {
//     }
// 
//     public void setResult(T result) {
//         this.result=result;
//         notifyListeners();
//         semaphore.release(Integer.MAX_VALUE - semaphore.availablePermits());
//     }
//     public Object getResult(long timeout,TimeUnit unit){
//         try {
//             if(!semaphore.tryAcquire(timeout,unit)){
//                 throw new RuntimeException("time out");
//             }
//         }catch (InterruptedException e){
//             throw new RuntimeException();
//         }
//         if(cause != null)
//             return null;
//         return result;
//     }
//     public void addInvokerListener(InvokeListener<T> listener){
//         this.listeners.add(listener);
//     }
//     private void notifyListeners(){
//         for(InvokeListener<T> listener : listeners){
//             listener.onResponse(result);
//         }
//     }
// }
// 
// 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;
//     }
// }


import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ConnectTimeoutException;
import model.InvokeFuture;
import model.StormResponse;
package producer.netty;


/**
 * Created by yang on 16-11-22.
 */
public class StormHandler extends ChannelInboundHandlerAdapter{
    private StormProducerConnection connect;
    private Throwable cause;
    private ConnectListener listener;
    public StormHandler(){

    }
    public StormHandler(StormProducerConnection conn){
        this.connect = conn;
    }
    public StormHandler(StormProducerConnection conn, ConnectListener listener) {
        this.connect = conn;
        this.listener = listener;
    }
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception{
        super.channelActive(ctx);
        System.out.println("connected on server: "+ ctx.channel().remoteAddress().toString());
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
       // System.out.println("channelRead");
// Path: src/main/java/com/ystruct.middleware/stormmq/model/InvokeFuture.java
// public class InvokeFuture<T> {
//     private Semaphore semaphore = new Semaphore(0);
//     private T result;
//     private List<InvokeListener<T>> listeners = new ArrayList<InvokeListener<T>>();
//     private String requestId;
//     private Throwable cause;
//     public void setCause(Throwable cause){
//         this.cause = cause;
//         notifyListeners();
//         semaphore.release(Integer.MAX_VALUE - semaphore.availablePermits());
// 
//     }
//     public String getRequestId() {
//         return requestId;
//     }
// 
//     public void setRequestId(String requestId) {
//         this.requestId = requestId;
//     }
//     public InvokeFuture() {
//     }
// 
//     public void setResult(T result) {
//         this.result=result;
//         notifyListeners();
//         semaphore.release(Integer.MAX_VALUE - semaphore.availablePermits());
//     }
//     public Object getResult(long timeout,TimeUnit unit){
//         try {
//             if(!semaphore.tryAcquire(timeout,unit)){
//                 throw new RuntimeException("time out");
//             }
//         }catch (InterruptedException e){
//             throw new RuntimeException();
//         }
//         if(cause != null)
//             return null;
//         return result;
//     }
//     public void addInvokerListener(InvokeListener<T> listener){
//         this.listeners.add(listener);
//     }
//     private void notifyListeners(){
//         for(InvokeListener<T> listener : listeners){
//             listener.onResponse(result);
//         }
//     }
// }
// 
// 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/producer/netty/StormHandler.java
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ConnectTimeoutException;
import model.InvokeFuture;
import model.StormResponse;

package producer.netty;


/**
 * Created by yang on 16-11-22.
 */
public class StormHandler extends ChannelInboundHandlerAdapter{
    private StormProducerConnection connect;
    private Throwable cause;
    private ConnectListener listener;
    public StormHandler(){

    }
    public StormHandler(StormProducerConnection conn){
        this.connect = conn;
    }
    public StormHandler(StormProducerConnection conn, ConnectListener listener) {
        this.connect = conn;
        this.listener = listener;
    }
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception{
        super.channelActive(ctx);
        System.out.println("connected on server: "+ ctx.channel().remoteAddress().toString());
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
       // System.out.println("channelRead");
        StormResponse response = (StormResponse)msg;