java - Observable と Emit の連鎖 / 元の Emit を Subscribe 呼び出しに渡す

Question

メッセージを消費して保存し、成功または失敗を返信するユースケースがあります。mongo 挿入は Observable を返すので、flatmap を使用して連鎖できます。問題は、挿入オブザーバブルが挿入の結果を発行することですが、返信するには最初のオブザーバブルから発行された元のメッセージが必要です。したがって、これを機能させるために、最初の Observable のサブスクライブ内で挿入を実行し、2 番目のサブスクライブ内で応答しています。

flatmap のようなある種の演算子を使用して、より反応的な方法でこれを達成したいと考えていました。演算子のリストを検索しましたが、探しているものが何も見つかりません。

eb.consumer("persister.save.event").toObservable()
    .subscribe(msg -> {
        mongo.insertObservable("event", (JsonObject) msg.body())
            .subscribe(
                res -> msg.reply(new JsonObject().put("success", true)),
                error -> msg.fail(500, "failed to save event"));
            });

上記のコードはそれを行うべき方法ですか、それともより良いアプローチがありますか? 2 つのサブスクライブは適切ではありません。