ReactiveArchitecture_NowPlayingInteractor_SeaOfAsync.java

public Observable<Result> processAction(Observable<Action> actions) {
    return actions.compose(transformActionIntoResults);
}

transformActionIntoResults = upstream -> upstream.publish(new Function<Observable<Action>, ObservableSource<Result>>() {
    @Override
    public ObservableSource<Result> apply(Observable<Action> actionObservable) throws Exception {                
        return Observable.merge(
            actionObservable.ofType(ScrollAction.class).compose(transformScrollActionToScrollResult),
            actionObservable.ofType(***Action.class).compose(transform***ActionToRestoreResult)
        );
   }
});