java - RxJava2-重用多个Rx链的运算符序列

我有以下Rx链：

compositeDisposable.add(manager.getObservable()
            .map(objects -> modelMapper.map(objects))
            .map(modelObjects -> {
                cache.save(modelObjects);
                return modelObjects ;
            })
            .flatMapIterable(modelObjects -> modelObjects)
            .sorted(objectComparator)
            .toList()
            .map(modelObjects -> viewModelMapper.map(modelObjects))
            .subscribe(this::onObjectsLoaded));

我想将运算符flatMapIterable(modelObjects -> modelObjects)，sorted(objectComparator)和toList()提取到一个单独的方法中，可以在多个Rx链中重复使用以对对象进行排序，因此该链看起来像这样：

compositeDisposable.add(manager.getObservable()
            .map(objects -> modelMapper.map(objects))
            .map(modelObjects -> {
                cache.save(modelObjects );
                return modelObjects ;
            })
            .compose(sortObjects())
            .map(modelObjects -> viewModelMapper.map(modelObjects))
            .subscribe(this::onObjectsLoaded));

是否可以创建这样的方法？
谢谢！

最佳答案

定义一个采用Observable并返回Observable的方法：

static <T> Observable<List<T>> sortListItem(Observable<List<T>> source,
        Comparator<? super T> comparator) {
    return source.flatMapIterable(v -> v)
                .toSortedList(comparator)
                .toObservable();
}

compositeDisposable.add(manager.getObservable()
        .map(objects -> modelMapper.map(objects))
        .map(modelObjects -> {
            cache.save(modelObjects);
            return modelObjects ;
        })
        .compose(ThisClass::sortListItem)
        .map(modelObjects -> viewModelMapper.map(modelObjects))
        .subscribe(this::onObjectsLoaded));