Как прослушать изменения и затем выполнить действие в Rx Android?

Question

 public Flowable<ViewModel> perform(
            Flowable<PaginationParam> paginationSource) {
        return Flowable.combineLatest(
                        mConnectionState.getConnectionState(),
                        mSessionInfo,
                        paginationSource,
                        Holder::combine)
                .flatMap(this::toViewModel); 
}

Теперь я хочу запустить execute (), когда я нахожусь на второй странице, поэтому я хочу прослушать изменение пагинации, а когда это не первая страница (0), то вызвать call execute ().

final Flowable<PaginationParam > paginationSource =
                mPagination.observe(this).filter(it -> it.getOffset() != 0);

Я не знаю, как их объединить .?

Answer 1

Вы можете изменить perform функцию:

public Flowable<ViewModel> perform(PaginationParam paginationParam) {
    return Flowable.combineLatest(
                    mConnectionState.getConnectionState(),
                    mSessionInfo,
                    {state, info -> Holder.combine(state, info, paginationParam)})
            .flatMap(this::toViewModel); 

}

и чем ее использовать:

paginationSource.flatMap( it -> perform(it))
            .subscribe(...)

Answer 2

perform(paginationSource)
.observeOn(AndroidSchedulers.mainThread())
.subscribe(viewModel -> {
 // doAction
}, ex-> {
 // Error
}, ()-> {
 // Completed, will not emit new action
})