Q:

How to compose streams\Observables correctly

I have conceptual problem understanding how to compose between streams\Observables which have different return type.
Here is a draft method I'm trying to code:
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())
                );

    });
}

The flow is as follow:
1. emit 2 Observable using .zip method.
2. do some logic on the return value of streams and if it results in error-code --> put it in list and return it so "subscribe" can return the error to user.
3. if no error, emit another "update" method using flatMap() - and this is where I have my problem.
4. eventually, all the results should be processed in "subscribe" because this is the point I acknowledge the user about his request.
Hope it's clear enough...
by the way, I'm trying to learn rxJava, but it's very hard I find there are enough\good sources - can someone recommend to me the best way to learn it?? I trying looking at tutorials at Youtube, Wikipedia, Github...most of them teaches using Scala and other scripting languages - couldn't find anything in Java.
Thank you for everyone that put the effort trying understand it!!

A:

I think you were almost there, but try breaking down the code inside your .zip lambda into smaller Rx operations. For example:
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);
            });
    })

The following classes are used for passing intermediate and final results:
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;
    }
}

Your subscriber then uses the packaged result similar to what you are already doing. 
.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())
            );
});

By using the intermediate and final results classes instead of passing around your results in a List<Object> your code is much more forgiving to changes and the compiler will type check everything for you.

