Sample.kt

    private fun calculateTeamMembership(): Observable<Timed<Map<User, Density>>> {
        return Observable.combineLatest(
                connectionProvider.connections(),
                repositoriesStorage.selectedRepositories(),
                BiFunction<BitbucketConnection, List<Repository>, Observable<Timed<Map<User, Density>>>> {
                    (userName, _, api, token), repositories ->
                    return@BiFunction  Observable.fromIterable(repositories)
                            .flatMap { (slug, _, project) ->
                                BitbucketApi.queryPaged { start ->
                                    api.getPullRequests(token, project.key, slug, start,
                                            status = STATUS_ALL, limit = PAGE_LIMIT)
                                }
                                        .subscribeOn(Schedulers.io())
                                        .take(PAGES_PER_REPOSITORY)
                                        .onErrorResumeNext(connectionProvider
                                                .handleNetworkError(ReviewersPresenter::class.java.simpleName))
                            }.reduce { t1, t2 -> t1 + t2 }.toObservable()
                            .compose(intoTeamMembershipOf(userName))
                            .switchIfEmpty(Observable.just(mapOf()))
                            .timestamp()
                }
        ).switchMap { stream -> stream }
    }