Chaining Observable＆Emitting/Passing Original Emit訂閱呼叫

Question

我有一個使用案例，我在使用消息，保存消息，然後回覆成功或失敗。 mongo插入返回一個Observable，所以我可以使用flatmap進行鏈接。問題是Insert Observable發出插入的結果，但我需要從第一個observable發出的原始消息進行回覆。因此，爲了做到這一點，我在第一個Observable的訂閱中運行插入，並在第二個訂閱內部進行回覆。

我一直希望用平面圖等某種操作符以更具反應性的方式完成此操作。我搜索了運營商名單，沒有找到我要找的東西。

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

上述代碼是應該完成的方式還是有更好的方法？這兩個訂閱者感覺不對。

Answer 1

這裏是什麼可以做，以避免兩個用戶：

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

的竅門是map您蒙戈結果想msg的flatMap內。