rx-java2 - RxJava を使用して Vertx JsonArray をトラバースする

Question

着信メッセージを消費するバーチクルがあります。各メッセージは、頂点JsonObjectを含む頂点になりJsonArrayます。この配列の各要素に対してロジックを実行したい。ロジック自体は別の項目に含まれています。この 2 番目の垂直は を使用しrxVertxます。複数のコンシューマーを定義し、それぞれが個別のメソッドに委譲し、そのすべてがObservable.

私の質問は次のとおりです。

1. の各要素をトラバースしますJsonArray
2. で動作するコンシューマーに各要素を渡しますObservables。

最初の項目では、次のことを試しました:

EventBus eb = rxVertx.eventBus();
JsonArray array= incomingMessage.getJsonArray(KEY);
List<Object> list  = array.getList();
Observable<Object> observable = Observable.fromArray(list);

observable.flatMapSingle(s -> {
      eb.rxSend(SECOND_VERTICLE_ADDRESS, s);
   }).subscribe();

次の理由により、への呼び出しflatMapSingleはコンパイルされません。

The method flatMapSingle(Function<? super Object,? extends SingleSource<? extends R>>) in the type Observable<Object> is not applicable for the arguments ((<no type> s) -> {})

これを行う正しい方法は何ですか？どうもありがとう