如何在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？我甚至需要结束它吗？

Answer 1

我认为你可以执行类似以下代码的操作。

主要思想是您不直接使用客户端HttpClientRequest获得的数据Vertx。end()相反，您创建另一个 Flowable，它将在收到第一个订阅后立即调用。

例如，在这里，您可以通过一对自定义方法获取请求：在本例中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() );
    }

}