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

}