如何处理Observable发出的项目以访问另一个值？

Question

我需要执行一个异步call_1，赶上它的Observable reply_1，然后再做一个异步call_2，并在处理它的reply_2时我还需要访问reply_1。

我已经试过类似：

public rx.Observable<Game> findGame(long templateId, GameModelType game_model, GameStateType state) { 

rx.Observable<RxMessage<byte[]>> ebs = context.getGameTemplate(templateId); 

return context.findGame(templateId, state) // findGame returns rx.Observable<RxMessage<byte[]>> 

    .flatMap(new Func1<RxMessage<byte[]>, rx.Observable<Game>>() { 

     @Override 
     public Observable<Game> call(RxMessage<byte[]> gameRawReply) { 

      Game game = null; 

      switch(game_model) { 

       case SINGLE: { 

        ebs.subscribe(new Action1<RxMessage<byte[]>>() { 

         @Override 
         public void call(RxMessage<byte[]> t1) { 

          game = singleGames.get(0); 

         } 
        }); 
       } 
      } 

      return rx.Observable.from(game); 
     } 
    }); 
} 

我还在编写本方法的game因为final问题的问题。

这是正确的方式来处理这个问题，或者有一种很自然的方式来完成我想要的。

Answer 1

如果我理解你想要正确地做什么，我想解决这个自然的方式将zip：

你有两个观测量异步发出他们的研究结果，即ebs和context.findGame(...)返回值。

你可以做这样的事情结合自己的结果：

public rx.Observable<Game> findGame(long templateId, GameModelType game_model, GameStateType state) { 

    rx.Observable<RxMessage<byte[]>> ebs = context.getGameTemplate(templateId); 
    rx.Observable<RxMessage<byte[]>> gameObs = context.findGame(templateId, state); 

    return Observable.zip(gameObs, ebs, new Func2<RxMessage<byte[]>, RxMessage<byte[]>, Game>() { 

     @Override 
     public Game call(RxMessage<byte[]> gameRawReply, RxMessage<byte[]> t1) { 

      Game game = null; 

      switch(game_model) { 
       case SINGLE: { 
        game = singleGames.get(0); 
       } 
      } 

      return game; 
     } 
    }); 
} 

的FUNC2 - 拉链的第三个参数 - 将被调用时源观测量的两个已经号召其onNext。它将用于将它们发出的值组合成一个新的Game类型的值，并将这些值发射给Observable的用户。

编辑：我是从Observable<Game>只是游戏改变（）调用的返回一些详细信息...

注意。否则，zip的结果不会是Observable<Game>，而是Observable<Observable<Game>>。与地图和flatMap不同的是，在rx中只有zip - 没有flatZip。但是因为你总是希望为每一对输入项目（一个来自ebs，一个来自gameObs）发出一个游戏，这在这种情况下不成问题。

另外，呼叫（）的FUNC2的现在可以进一步简化为：

@Override 
public Game call(RxMessage<byte[]> gameRawReply, RxMessage<byte[]> t1) { 

    switch(game_model) { 
     case SINGLE: { 
      return singleGames.get(0); 
     } 
    } 
}