如何正确组合流可观察对象

Question

我在理解如何在具有不同返回类型的流\Observables之间进行组合方面存在概念性问题。

这是一个我正在尝试编码的方法草案：

public void findSeat() {
    rx.Observable<GameObject> userObs = context.getUser();
    rx.Observable<ActiveGame> gameObs = context.findGame();
    rx.Observable.zip(userObs, gameObs, (userObj, game) -> {
        User user = ...;
        final List<Object> results = new ArrayList<Object>(3); 
        if(userObj.getStatus() != ErrorCodes.STATUS_OK) {
            results.add(-1);
            return results;
        }
        ...
        ...
        //***********************************
        // THE PROBLEM IS HERE: 
        // "context.getActiveGameManager().updateGame(game)" returns Observable<GameOBject> and not List<Object> like .zip() expects.
        // because of that I cannot do:
        // "return context.getActiveGameManager().updateGame(game);"
        // How can I do this convertion from Observable<GameObject> to List<Object>
        //************************************
        context.getActiveGameManager().updateGame(game)
            .map((gameObj) -> {
                if(gameObj.getStatus() != ErrorCodes.STATUS_OK) {
                    results.add(-2);
                    return (Observable<? extends Object>) results;
                }
                results.add(ErrorCodes.STATUS_OK);
                results.add(user);
                results.add(gameObj);
                return gameObs;
        });
        return Observable.empty();
    }).subscribe((results) -> {
        int status = (int) results.get(0);
        User user = (User) results.get(1);
        ActiveGame game = (ActiveGame) results.get(2);

        replyObj.reply(new JsonObject()
                    .putString("action", CommandActions.FIND_SEAT)
                    .putNumber("status", status);
                    .putNumber("game_id", game.getGameId())
                );
    });
}

流程如下：1.使用.zip方法发射2个Observable。2.对流的返回值进行一些逻辑运算，如果它导致错误代码-->，请将其放在列表中并返回，以便"订阅"可以将错误返回给用户。3.如果没有错误，使用flatMap（）发出另一个"update"方法——这就是我的问题所在。4.最终，所有结果都应该在"订阅"中处理，因为这是我向用户确认其请求的要点。

希望它足够清楚。。。

顺便说一句，我正在努力学习rxJava，但我很难找到足够多的好资源——有人能向我推荐学习它的最佳方法吗？？我试着在Youtube、维基百科、Github上看教程。。。他们中的大多数人教授使用Scala和其他脚本语言——在Java中找不到任何东西。

感谢所有努力理解它的人！！

Answer 1

我想你已经差不多了，但试着把.zip lambda中的代码分解成更小的Rx操作。例如：

rx.Observable
    .zip(userObs, gameObs, (userObj, game) -> {
        // Combine the user & game objects and pass them to the
        // next Rx operation.
        return new UserAndActiveGame(userObj, game);
    })
    .filter(userAndActiveGame -> {
        // Remove this filter if you want errors to make it to the subscriber.
        return userAndActiveGame.getUserObj().getStatus() == ErrorCodes.STATUS_OK;
    })
    .flatMap(userAndActiveGame -> {
        // Remove this check if you filter errors above.
        if (userAndActiveGame.getUserObj().getStatus() != ErrorCodes.STATUS_OK) {
            return Observable.just(new FindSeatResult(-1));
        }
        return context.getActiveGameManager().updateGame(userAndActiveGame.getGame())
            .map(gameObj -> {
                if (gameObj.getStatus() != ErrorCodes.STATUS_OK) {
                    return new FindSeatResult(-2);
                }
                User user =...; // Whatever you are doing to get this in your example code.
                return new FindSeatResult(ErrorCodes.STATUS_OK, user, gameObj);
            });
    })

以下类别用于传递中间和最终结果：

private class UserAndActiveGame {
    private final GameObject userObj;
    private final ActiveGame game;
    public UserAndActiveGame(GameObject userObj, ActiveGame game) {
        this.userObj = userObj;
        this.game = game;
    }
    public GameObject getUserObj() {
        return userObj;
    }
    public ActiveGame getGame() {
        return game;
    }
}
private class FindSeatResult {
    private final int status;
    private final User user;
    private final ActiveGame game;
    public FindSeatResult(int status) {
        this(status, null, null);
    }
    public FindSeatResult(int status, User user, ActiveGame game) {
        this.status = status;
        this.user = user;
        this.game = game;
    }
    public User getUser() {
        return user;
    }
    public int getStatus() {
        return status;
    }
    public ActiveGame getGame() {
        return game;
    }
}

然后，您的订阅者使用与您已经在做的类似的打包结果。

.subscribe((results) -> {
    // You don't need this if you filter errors above.
    if (findSeatResult.getStatus() == -1) {
        return;
    }
    int status = findSeatResult.getStatus();
    User user = findSeatResult.getUser();
    ActiveGame game = findSeatResult.getGame();
    replyObj.reply(new JsonObject()
                .putString("action", CommandActions.FIND_SEAT)
                .putNumber("status", status);
                .putNumber("game_id", game.getGameId())
            );
});

通过使用中间和最终结果类，而不是在List<Object>中传递结果，您的代码对更改更宽容，编译器将为您键入检查所有内容。