java - Rx java 运算符；将数据流封装到自定义 Operator 中

Question

假设我正在以一种非常具体的方式观察一个可观察对象。

    resultObservable = anotherObservable.filter(~Filter code~).take(15);  

我想创建一个自定义运算符，它结合了两个预定义的运算符，例如过滤器和取值。这样它的行为就像

    resultObservable = anotherObservable.lift(new FilterAndTake(15));  

或者……

    resultObservable = anotherObservable.FilterAndTake(15);  

到目前为止，我很乐意编写可以执行此操作的非常具体的运算符。我可以举起那个接线员。

但是，鉴于我目前对 rx java 的了解有限，每次我需要在自定义运算符中使用它时，这将涉及重新编写 take 和 filter 功能。

这样做很好，但我宁愿重复使用由开源社区维护的现有运算符，以及我创建的回收运算符。

还有一点告诉我，我对运营商和订阅者缺乏足够的了解。

有人可以推荐不是 rx-java 文档的教程吗？
我这样说是因为，虽然文档解释了一般概念，但它隔离了概念和它们功能的一般上下文，没有留下任何示例来激发更强大的 RX java 应用程序。

基本上

我正在尝试将自定义数据流封装到代表性运算符中。是否存在此功能？

Answer 1

我不知道组成 Operator 对象的一些特殊函数(或糖)。但是您可以简单地创建一个新的 Operator 来组合现有的 Operator。以下是 FilterAndTake 运算符的工作示例:

public class FilterAndTake<T> implements Observable.Operator<T, T> {

    private OperatorFilter<T> filter;
    private OperatorTake<T> take;

    public FilterAndTake(Func1<? super T, Boolean> predicate, int n) {
        this.filter = new OperatorFilter<T>(predicate);
        this.take = new OperatorTake<T>(n);
    }

    @Override
    public Subscriber<? super T> call(final Subscriber<? super T> child) {
        return filter.call(take.call(child));
    }
}

然后你可以按如下方式使用它:

public static void main(String[] args) {
    Observable<Integer> xs = Observable.range(1, 8);

    Func1<Integer, Boolean> predicate = new Func1<Integer, Boolean>() {
        @Override
        public Boolean call(Integer x) {
            return x % 2 == 0;
        }
    };

    Action1<Integer> action = new Action1<Integer>() {
        @Override
        public void call(Integer x) {
            System.out.println("> " + x);
        }
    };

    xs.lift(new FilterAndTake<Integer>(predicate, 2)).subscribe(action);
}