観測可能/送信/通過元の発信を購読するには

Question

私はメッセージを消費し、保存してから、成功または失敗を返すユースケースを持っています。 mongoインサートはObservableを返すので、私はフラットマップを使ってチェーンすることができます。問題はinsert Observableが挿入の結果を出力することですが、最初のobservableから元のMessage emitを返す必要があります。したがって、この作業を行うために、最初のObservableを購読し、2番目の購読の中で返信することでインサートを実行しています。

フラットマップのようなオペレータでより反応的な方法でこれを実現したいと考えていました。私は演算子のリストを検索し、私が探しているものは何も出てこない。

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人の購読者は正しく感じません。

Answer 1

はここで2人の加入者を避けるために何ができるかです：

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

トリックはflatMap内msgを望んでいたために、あなたのmongo結果をmapすることです。