Повторное использование части потока RxJava

Question

Вероятно, это вопрос rx newbee.

Если у меня есть два потока RX, которые имеют некоторые общие части, есть ли возможность извлечь и использовать их повторно?

Вот пример псевдокода:

someManager.getInfo(id) returns a Single<SometInfo>

Это веселье принимает идентификатор и должен вернуть только один со статусом

fun getStatus(id: String): Single<Status> =
    someManager.getInfo(id)
        .flattenAsObservable { it.items }
        .map { getSomeProp(it) } // from here
        .toList()
        .map { getStatus(it) }
        .doOnSuccess { getStatusRelay(id).accept(it) } // until here it's common

и этот не берет идентификатор, но уже является объектом и должен проверить, если статус в порядке

fun isStatusOk(info: SomeInfo): Single<Boolean> =
    Observable.fromIterable(info.items)
        .map { getSomeProp(it) } // from here
        .toList()
        .map { getStatus(it) }
        .doOnSuccess { getStatusRelay(id).accept(it) } // until here it's common
        .map { it == Status.OK }

заранее спасибо

Answer 1

Вы также можете использовать функцию расширения

fun Observable<Item>.convertToSingeStatus(): Single<Status> =
        this.map { getSomeProp(it) }
            .toList()
            .map { getStatus(it) }
            .doOnSuccess { getStatusRelay(id).accept(it) }

А потом

fun getStatus(id: String): Single<Status> =
someManager.getInfo(id)
    .convertToSingeStatus()
    .doOnSuccess { getStatusRelay(id).accept(it) }

Answer 2

Феникс Ван предложил Трансформеры. Я попробовал их, и после некоторых экспериментов, кажется, я получил это работает.

Я создал SingleTransformer:

private fun getStatus() =
    SingleTransformer<SomeInfo, Status> {
        it.flatMap {
            Observable.fromIterable(it.items)
                .map { getSomeProp(it) }
                .toList()
                .map { getStatus(it) }
        }
    }

И тогда я могу рефакторинг

fun getStatus(id: String): Single<Status> =
    someManager.getInfo(id)
        .compose(getStatus())
        .doOnSuccess { getStatusRelay(id).accept(it) }

и это тоже

fun isStatusOk(info: SomeInfo): Single<Boolean> =
    Single.just(info)
        .compose(getStatus())
        .doOnSuccess { getStatusRelay(id).accept(it) }
        .map { it == Status.OK }

Надеюсь, это поможет кому-то еще.