Cliente HTTP asíncrono con Netty

Question

Soy nuevo en netty y todavía estoy luchando para encontrar mi camino. Estoy buscando crear un cliente http que funcione de manera asíncrona. Los ejemplos netos de http solo muestran cómo esperar las operaciones IO, y no cómo usar addListener, así que he estado tratando de resolver esto durante los últimos días.

Estoy tratando de crear una clase de solicitud que manejará todos los diferentes estados de una solicitud, desde la conexión, el envío de los datos, el manejo de la respuesta y el cierre de la conexión. Para hacer eso, mi clase extiende SimpleChannelUpstreamHandler e implementa ChannelFutureListener. Uso un ChannelPipelineFactory que agrega la (esta) instancia la clase (como SimpleChannelUpstreamHandler) a la canalización como controlador.

La conexión se crea de esta manera:

this.state = State.Connecting; 
this.clientBootstrap.connect(this.address).addListener(this); 

Entonces el operationComplete método:

@Override 
public void operationComplete(ChannelFuture future) throws Exception { 
    State oldState = this.state; 

    if (!future.isSuccess()) { 
     this.status = Status.Failed; 
     future.getChannel().disconnect().addListener(this); 
    } 
    else if (future.isCancelled()) { 
     this.status = Status.Canceled; 
     future.getChannel().disconnect().addListener(this); 
    } 
    else switch (this.state) { 
     case Connecting: 
      this.state = State.Sending; 
      Channel channel = future.getChannel(); 
      channel.write(this.createRequest()).addListener(this); 
      break; 

     case Sending: 
      this.state = State.Disconnecting; 
      future.getChannel().disconnect().addListener(this); 
      break; 

     case Disconnecting: 
      this.state = State.Closing; 
      future.getChannel().close().addListener(this); 
      break; 

     case Closing: 
      this.state = State.Finished; 
      break; 
    } 
    System.out.println("request operationComplete start state: " + oldState + ", end state: " + this.state + ", status: " + this.status); 
} 

private HttpRequest createRequest() { 
    String url = this.url.toString(); 

    HttpRequest request = new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, url); 
    request.setHeader(HttpHeaders.Names.HOST, this.url.getHost()); 
    request.setHeader(HttpHeaders.Names.CONNECTION, HttpHeaders.Values.CLOSE); 
    request.setHeader(HttpHeaders.Names.ACCEPT_ENCODING, HttpHeaders.Values.GZIP); 

    return request; 
} 

La clase también anula el messageReceived método:

@Override 
public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception { 
    System.out.println("messageReceived"); 
    HttpResponse response = (HttpResponse) e.getMessage(); 

    ChannelBuffer content = response.getContent(); 
    if (content.readable()) { 
     System.out.println("CONTENT: " + content.toString(CharsetUtil.UTF_8)); 
    } 
} 

El problema es que me da este resultado:

request operationComplete start state: Connecting, end state: Sending, status: Unknown 
request operationComplete start state: Sending, end state: Disconnecting, status: Unknown 
request operationComplete start state: Closing, end state: Finished, status: Unknown 
request operationComplete start state: Disconnecting, end state: Finished, status: Unknown 

Como se puede ver la messageReceived del no está siendo ejecutada por alguna razón, a pesar de que la fábrica de tuberías añade la instancia de esta clase a la tubería.

¿Alguna idea de lo que me falta aquí? Gracias.

---

Editar

logré finalmente este trabajo gracias a la ayuda de @JestanNirojan, en caso de que alguien va a estar interesado en la solución:

public class ClientRequest extends SimpleChannelUpstreamHandler { 

    .... 

    public void connect() { 
     this.state = State.Connecting; 
     System.out.println(this.state); 
     this.clientBootstrap.connect(this.address); 
    } 

    @Override 
    public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception { 
     this.state = State.Sending; 
     System.out.println(this.state); 
     ctx.getChannel().write(this.createRequest()); 
    } 

    @Override 
    public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception { 
     HttpResponse response = (HttpResponse) e.getMessage(); 

     ChannelBuffer content = response.getContent(); 
     if (content.readable()) { 
      System.out.println("CONTENT: " + content.toString(CharsetUtil.UTF_8)); 
     } 

     this.state = State.Disconnecting; 
     System.out.println(this.state); 
    } 

    @Override 
    public void channelDisconnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception { 
     this.state = State.Closing; 
     System.out.println(this.state); 
    } 

    @Override 
    public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception { 
     this.state = State.Finished; 
     System.out.println(this.state); 
    } 

    private HttpRequest createRequest() { 
     String url = this.url.toString(); 

     HttpRequest request = new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, url); 
     request.setHeader(HttpHeaders.Names.HOST, this.url.getHost()); 
     request.setHeader(HttpHeaders.Names.CONNECTION, HttpHeaders.Values.CLOSE); 
     request.setHeader(HttpHeaders.Names.ACCEPT_ENCODING, HttpHeaders.Values.GZIP); 

     return request; 
    } 
} 

Answer 1

Está utilizando un ChannelFutureListener a hacer todas las operaciones en el canal (que es malo), y el oyente futuro se ejecutará justo después de llamar a esas operaciones de canal.

El problema es que, después de enviar el mensaje, el canal se desconecta inmediatamente y el controlador no puede recibir el mensaje de respuesta que viene después.

 ........ 
    case Sending: 
     this.state = State.Disconnecting; 
     future.getChannel().disconnect().addListener(this); 
     break; 
     ........ 

no debe bloquear el hilo futuro del canal en absoluto. El mejor enfoque es extender los métodos de SimpleChannelUpstreamHandler

channelConnected(..) {} 
    messageReceived(..) {} 
    channelDisconnected(..) {} 

y reaccionar a esos eventos. también puedes mantener el estado en ese controlador.