Sample2.kt

    private fun pullRequests(view: T): ConnectableObservable<Pair<List<PullRequest>, String>> {
        return Observable.combineLatest(
                view.intentPullToRefresh().startWith { Object() },
                connectionProvider.connections(),
                BiFunction<Any, BitbucketConnection, BitbucketConnection> { _, conn -> conn }
        ).switchMap { (_, serverUrl, api, token) ->
            repositoriesStorage.selectedRepositories()
                    .switchMap { list ->
                        Observable.fromIterable(list)
                                .flatMap { (slug, _, project) ->
                                    BitbucketApi.queryPaged { start ->
                                        api.getPullRequests(token, project.key, slug, start, avatarSize()) }
                                            .subscribeOn(Schedulers.io())
                                            .onErrorResumeNext(connectionProvider.handleNetworkError(
                                                    BasePullRequestsPresenter::class.java.simpleName))
                                }.reduce { t1, t2 -> t1 + t2 }.map { it to serverUrl }.toObservable()
                                .observeOn(AndroidSchedulers.mainThread())
                                .doOnComplete { view.onLoadingCompleted() }
                                .switchIfEmpty(Observable.just(NULL_PULL_REQUESTS to serverUrl))
                    }
        }
                .doOnSubscribe { view.onSelfLoadingStarted() }
                .publish()
    }