java - 如何处理由 Observable 发出的项目并访问另一个 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);
        }
    });
}

由于游戏的final问题，我在编译此方法时仍然遇到问题。

这是解决这个问题的正确方法吗？还是有一种更自然的方法来完成我正在尝试的事情。

Answer 1

如果我正确理解你想要做什么，我认为解决这个问题的自然方法是 zip :

您有两个异步发出结果的 Observable，即 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 - zip 的第三个参数 - 将在源 Observables 的两个调用它们的 onNext 时被调用。 。它将用于将它们发出的值组合为 Game 类型的新值，并将该值发送给生成的 Observable 的订阅者。

编辑:更多信息...

请注意，我将 call() 的返回值从 Observable<Game> 更改为只是游戏。否则 zip 的结果就不会是 Observable<Game>但是Observable<Observable<Game>> 。与 map 和 flatMap 不同，rx 中只有 zip - 没有 flatZip。但由于您总是希望为每对输入项(一个来自 ebs，一个来自 gameObs)发出一个游戏，所以在这种情况下这不是问题。

此外，Func2 的 call() 现在可以进一步简化为:

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

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