Vert.x - RxJava - zip observables

Question

我正在尝试使用Vert.x和RxJava压缩到可观察量。我不知道我是否误解了什么，或者这只是某种错误。这是代码。

public class BusVerticle extends Verticle {
    public void start() {
        final RxVertx rxVertx = new RxVertx(vertx);
        Observable<RxMessage<JsonObject>> bus = rxVertx.eventBus().registerHandler("busName");
        Observable<RxHttpClientResponse> httpResponse = bus.mapMany(new Func1<RxMessage<JsonObject>, Observable<RxHttpClientResponse>>() {
            public Observable<RxHttpClientResponse> call(RxMessage<JsonObject> rxMessage) {
                RxHttpClient rxHttpClient = rxVertx.createHttpClient();
                rxHttpClient.coreHttpClient().setHost("localhost").setPort(80);
                return rxHttpClient.getNow("/uri");
            }
        });
        Observable<RxMessage<JsonObject>> zipObservable = Observable.zip(bus, httpResponse, new Func2<RxMessage<JsonObject>, RxHttpClientResponse, RxMessage<JsonObject>>() {
            public RxMessage<JsonObject> call(RxMessage<JsonObject> rxMessage, RxHttpClientResponse rxHttpClientResponse) {
                return rxMessage;
            }
        });
        zipObservable.subscribe(new Action1<RxMessage<JsonObject>>() {
            public void call(RxMessage<JsonObject> rxMessage) {
                rxMessage.reply();
            }
        });
    }
}

我想使用收到的消息中的信息发出 HTTP 请求，然后压缩两个可观察量，事件总线和 HTTP 响应，以便使用 HTTP 响应中的信息回复消息。

我发送的消息没有得到任何回复。

提前感谢！

Answer 1

我已经用解决方法解决了它。某种混合解决方案。

public class BusVerticle extends Verticle {
public void start() {
    final RxVertx rxVertx = new RxVertx(vertx);
    vertx.eventBus().registerHandler("busName", new Handler<Message<JsonObject>>() {
        public void handle(final Message<JsonObject> message) {
            RxHttpClient rxHttpClient = rxVertx.createHttpClient();
            rxHttpClient.coreHttpClient().setHost("localhost").setPort(80);
            Observable<RxHttpClientResponse> httpRequest = rxHttpClient.getNow("/uri");
            httpRequest.subscribe(new Action1<RxHttpClientResponse>() {
                public void call(RxHttpClientResponse response) {
                    container.logger().error(response.statusCode());
                    message.reply(new JsonObject().putString("status", "ok"));
                }
            });
        }
    });
}

}