java - 发出 2 个可观察量，处理结果，然后发出另一个

Question

我有以下任务要执行:

我需要发出 2 个可观察量(obs1 和 obs2)处理它们的结果，然后调用另一个可观察量(obs3)并处理其结果，如果可能的话，在处理 obs3 的结果时可以访问 obs1 和 obs2 的结果。

这是我的草稿代码，但不起作用，我该如何更改它。

public void executeFind(String session_id, long template_id, GameModelType game_model) {

   Observable<RxMessage<byte[]>> userObs = context.getUser(session_id);
   Observable<Game> gameObs = context.findGame(template_id, game_model, GameStateType.WAITING);

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

     @Override
     public GameObject call(RxMessage<byte[]> userRawReply, ActiveGame game) {

     ..
     ..
     return context.updateGame(game.getGameData())
            .subscribe(new Action1<GameObject>() {

                @Override
                public void call(GameObject updateReply) {
                    ..
                    ..

                }
            });

        return userReply; 
     }
   });
 }

这实际上不起作用 - 我可以编写一个代码，对每个 Observable 使用显式调用 .flatMap\subscribe ，但会导致许多嵌套调用，这显然是框架的不良使用。

解决这个问题的正确方法是什么？

谢谢!

编辑:

我发现这个解决方案可行，但我仍然想知道是否有一种“更干净”的方法来实现这一点:

public void executeFind(ReplyMessage<JsonObject> replyObj, String session_id, long template_id,     GameModelType game_model) throws CommandException {

    rx.Observable<GameObject> userObs = context.getUser(session_id);
    rx.Observable<Game> gameObs = context.findGame(template_id, game_model, GameStateType.WAITING);

    rx.Observable.zip(userObs, gameObs, new Func2<GameObject, Game, List<Object>>() {

        @Override
        public List<Object> call(GameObject userReply, Game game) {

            User user = ...;

            final List<Object> results = new ArrayList<Object>(3); 
            results.add(ErrorCodes.STATUS_OK);
            results.add(user);
            results.add(game);


            context.updateGame(game.getGameData()).subscribe(new Action1<GameObject>() {

                @Override
                public void call(GameObject updateReply) {

                    ...
                }

            });


            return results;
        }
    }).subscribe(new Action1<List<Object>>() {

        @Override
        public void call(List<Object> results) {

            int status = (int) results.get(0);
            User user = (User) results.get(1);
            Game game = (Game) results.get(2);

        }

    });
}

Answer 1

我会根据以下想法编写这个东西。如果与您的用例相关，可以将 map 替换为 flatMap。另请注意，我只使用了 Java 8 lambda 语法，但为了提高可读性，我强烈建议您为每个函数/操作使用简单且命名良好的方法(并将它们与方法引用一起使用)，因为这将提高代码的可理解性(这就是我们在mockito上所做的，但每个人都应该在自己的代码库中这样做)。

public void executeFind(ReplyMessage<JsonObject> reply_obj, String session_id, long template_id,     GameModelType game_model) throws CommandException {
    Observable<GameObject> userObs = context.getUser(session_id);
    Observable<Game> gameObs = context.findGame(template_id, game_model, GameStateType.WAITING);
    Observable.zip(userObs, gameObs, (userReply, game) -> {
        User user = ...;
        return GameOfUser.gameFound(game, user);
    }).map(gou -> { 
        context.updateGame(gou.gameData()).susbcribe(...);
        return gou;
    }).subscribe(gou -> ...);
}