rxjava eb

rxjavaEb.java

//返回一个Obserable,侦听发给heat-sensor的信息
Observable<Double> observable = vertx.eventBus().
    <Double>consumer("heat-sensor").
    bodyStream().
    toObservable();

//这个obserable持续1秒钟，并对数据进行map reduce操作
observable.
    buffer(1, TimeUnit.SECONDS).
    map(samples -> samples.
        stream().
        collect(Collectors.averagingDouble(d -> d))).
    subscribe(heat -> {
      //再发给另外一个consumer通过eb
      vertx.eventBus().send("news-feed", "Current heat is " + heat);
    });