Асинхронный HTTP-клиент с Netty - PullRequest
Новая
       22

Асинхронный HTTP-клиент с Netty

10 голосов
/ 17 марта 2012

Я новичок в нетти и все еще пытаюсь найти свой путь.Я ищу, чтобы создать http-клиент, который работает асинхронно.Примеры http для netty показывают только, как ждать операций ввода-вывода, а не как использовать addListener , и поэтому я пытался выяснить это в течение последних нескольких дней.

Я пытаюсь создать класс запроса, который будет обрабатывать все различные состояния запроса, начиная с соединения, отправки данных, обработки ответа и затем закрытия соединения.Для этого мой класс расширяет SimpleChannelUpstreamHandler и реализует ChannelFutureListener .Я использую ChannelPipelineFactory , который добавляет (this) экземпляр класса (как SimpleChannelUpstreamHandler ) к конвейеру в качестве обработчика.

Соединение создается следующим образом:

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

Затем метод operationComplete : 

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

Класс также переопределяет метод messageReceived : 

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

Проблема в том, что я получаю следующие выходные данные: 

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

Как вы можете видеть, messageReceived не выполняется по какой-то причине, хотя фабрика конвейеров добавляет экземплярэтого класса в конвейер.

Есть идеи, что я здесь скучаю?Спасибо.

Редактировать

Мне удалось, наконец, получить эту работу благодаря помощи @JestanNirojan, на случай, если кто-то заинтересуется решением: 

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

1 Ответ

3 голосов
/ 17 марта 2012

Вы используете ChannelFutureListener для выполнения всех операций в канале (что плохо), и будущий слушатель будет выполнен сразу после вызова этих операций канала.

Проблема в том, что после отправки сообщения канал немедленно отключается, и обработчик не может получить ответное сообщение, которое приходит позже. 

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

Вы вообще не должны блокировать канал будущего потока. Лучший подход - это расширение SimpleChannelUpstreamHandler

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

методы и реагировать на эти события. вы также можете сохранить состояние в этом обработчике.

...