Asynchroniczny klient HTTP z Netty

Question

Jestem nowy dla netty i wciąż mam problemy z odnalezieniem swojej drogi. Szukam utworzyć klienta http, który działa asynchronicznie. Netty przykłady http tylko pokazują, jak czekać na operacje IO, a nie jak korzystać z addListener, i tak próbowałem to zrozumieć przez ostatnie kilka dni.

Próbuję utworzyć klasę żądań, która obsłuży wszystkie różne stany żądania, od łączenia, wysyłania danych, obsługi odpowiedzi, a następnie zamykania połączenia. Aby to zrobić, moja klasa rozszerza SimpleChannelUpstreamHandler i implementuje ChannelFutureListener. Używam ChannelPipelineFactory, który dodaje (tej) instancji klasę (jako SimpleChannelUpstreamHandler) do potoku jako program obsługi.

Połączenie jest stworzony tak:

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

wtedy operationComplete metody:

@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; 
} 

Klasa zasłania również MessageReceived metody:

@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)); 
    } 
} 

Problemem jest to, że otrzymuję ten wynik:

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 

Jak widać MessageReceived z poniższych nie jest wykonany z jakiegoś powodu, mimo że fabryka rurociąg dodaje instancję tej klasy do rurociągu.

Jakieś pomysły, których tu brakuje? Dzięki.

---

Edit

udało mi się w końcu dostać to pracę dzięki pomocy @JestanNirojan, w przypadku gdy ktoś będzie zainteresowany w roztworze:

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

używasz ChannelFutureListener do wykonaj wszystkie operacje na kanale (które są złe), a przyszły detektor zostanie wykonany zaraz po wywołaniu tych operacji kanału.

Problem polega na tym, że po wysłaniu wiadomości kanał jest natychmiast rozłączany, a przewodnik nie może otrzymać komunikatu odpowiedzi, który pojawia się później.

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

Nie należy w ogóle blokować przyszłego wątku kanału. Najlepsze podejście polega na rozszerzeniu metod i reakcji na te zdarzenia SimpleChannelUpstreamHandler. możesz także zachować stan w tym programie obsługi.