如何在RxJava Vert.x中结束链接的HTTP请求？

Question

如何在Rx Vert.X中结束链接请求？

 HttpClient client = Vertx.vertx().createHttpClient();
        HttpClientRequest request = client.request(HttpMethod.POST,
                "someURL")
                .putHeader("content-type", "application/x-www-form-urlencoded")
                .putHeader("content-length", Integer.toString(jsonData.length())).write(jsonData);
        request.toObservable().
                //flatmap HttpClientResponse -> Observable<Buffer>
                        flatMap(httpClientResponse -> { //something
                    return httpClientResponse.toObservable();
                }).
                        map(buffer -> {return buffer.toString()}).
                //flatmap data -> Observable<HttpClientResponse>
                        flatMap(postData -> client.request(HttpMethod.POST,
                        someURL")
                        .putHeader("content-type", "application/x-www-form-urlencoded")
                        .putHeader("content-length", Integer.toString(postData.length())).write(postData).toObservable()).
                //flatmap HttpClientResponse -> Observable<Buffer>
                        flatMap(httpClientResponse -> {
                    return httpClientResponse.toObservable();
                })......//other operators
request.end();

请注意，我对顶部请求使用了end()。那么如何结束在flatmap中的请求？我是否需要结束它？

- newprint

Answer 1

确保调用request.end()有多种方法。但如果有的话，我会查看Vert.x的文档或开源代码，以确定它是否为您调用end()。否则，可以尝试以下方法：

final HttpClientRequest request = ...
request.toObservable()
       .doOnUnsubscribe(new Action0() {
           @Override
           public void call() {
               request.end();
           }
       });

- Volodymyr Lykhonis

Answer 2

我认为你可以尝试以下代码。

主要思路是不直接使用由Vertx客户端获得的HttpClientRequest。相反，您可以创建另一个可流动对象，在第一次订阅时调用end()方法。

例如，在此处，您可以通过一对自定义方法（在本例中为request1()和request2()）获取请求。它们都使用doOnSubscribe()触发所需的end()方法。请阅读ReactiveX页面上的描述。

此示例使用vertx和reactivex，希望您能使用这个设置。

import io.reactivex.Flowable;
import io.vertx.core.http.HttpMethod;
import io.vertx.reactivex.core.Vertx;
import io.vertx.reactivex.core.buffer.Buffer;
import io.vertx.reactivex.core.http.HttpClient;
import io.vertx.reactivex.core.http.HttpClientRequest;
import io.vertx.reactivex.core.http.HttpClientResponse;
import org.junit.Test;

public class StackOverflow {

    @Test public void test(){

        Buffer jsonData = Buffer.buffer("..."); // the json data.

        HttpClient client = Vertx.vertx().createHttpClient(); // the vertx client.

        request1(client)
            .flatMap(httpClientResponse -> httpClientResponse.toFlowable())
            .map(buffer -> buffer.toString())
            .flatMap(postData -> request2(client, postData) )
            .forEach( httpResponse -> {
                // do something with returned data);
            });

    }

    private Flowable<HttpClientResponse> request1(HttpClient client) {
        HttpClientRequest request = client.request(HttpMethod.POST,"someURL");
        return request
                .toFlowable()
                .doOnSubscribe( subscription -> request.end() );
    }

    private Flowable<HttpClientResponse> request2(HttpClient client, String postData) {
        HttpClientRequest request = client.request(HttpMethod.POST,"someURL");
        // do something with postData
        return request
                .toFlowable()
                .doOnSubscribe( subscription -> request.end() );
    }

}

- danidemi