java - 如何组合相互依赖的 Observables 并获得一个包含每个值的对象？

Question

我正在学习 RxJava，并且想知道如何解决这个问题:我正在使用 Twitch API，我想获取流，并为每个流获取其游戏，最后保存在一个对象中并显示它。

这是我的代码:

twitchAPI.getStreamsObservable(TwitchAPI.CLIENT_ID)
                .flatMap(twitchStreams -> Observable.fromIterable(twitchStreams.getStreamList()))
                .filter(Stream -> Stream.getViewerCount() > 10000)
.subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(stream -> {
                    StreamMostViewed streamMostViewed = new StreamMostViewed();
                    streamMostViewed.setStreamName(stream.getTitle());
                    twitchAPI.getGame(stream.getGameId(), TwitchAPI.CLIENT_ID)
                            .subscribeOn(Schedulers.io())
                            .observeOn(AndroidSchedulers.mainThread())
                            .subscribe(game -> {
                                streamMostViewed.setGameName(game.getName());
                                view.showStream(streamMostViewed);
                            });
                });

问题是，首先，它获取所有流，然后获取游戏，但所有这些都是空的...

我的另一个解决方案是:

twitchAPI.getStreamsObservable(TwitchAPI.CLIENT_ID)
                .flatMap(twitchStreams -> Observable.fromIterable(twitchStreams.getStreamList()))
                .filter(Stream -> Stream.getViewerCount() > 10000)
 .flatMap(stream -> twitchAPI.getGame(stream.getGameId(),TwitchAPI.CLIENT_ID))
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(game -> {
                    game.getName();
//How could I get the stream object??
                });

但在订阅中我不知道如何获取流对象，以便保存在对象中并显示它。

谢谢!!

Answer 1

最优雅的方式是将流“拆分”为两部分。

streams = twitchAPI.getStreamsObservable(TwitchAPI.CLIENT_ID)
            .flatMap(twitchStreams -> Observable.fromIterable(twitchStreams.getStreamList()))
            .filter(Stream -> Stream.getViewerCount() > 10000)
            .replay()
            .autoConnect(2);

第一次订阅:

streams.subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(stream -> {
            // handle `stream`
        });

第二次订阅。

streams.flatMap(stream -> twitchAPI.getGame(stream.getGameId(),TwitchAPI.CLIENT_ID))
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(game -> {
            // do something with the `game`
        });