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

【问题标题】：Rx java Operators; Encapsulating data flow into custom OperatorsRx java 运算符；将数据流封装到自定义 Operator 中
【发布时间】：2014-09-29 16:58:06
【问题描述】：

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

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

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

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

或者……

    resultObservable = anotherObservable.FilterAndTake(15);

到目前为止，我对编写一个可以做到这一点的非常具体的运算符感到满意。我可以举起那个操作员。

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

这样做很好，但我宁愿重用由开源社区维护的预先存在的运算符，以及我创建的循环运算符。

还有一些事情告诉我，我对运营商和订户缺乏足够的了解。

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

非常重要

我正在尝试将自定义数据流封装到有代表性的运算符中。有这个功能吗？

【问题讨论】：

标签： java reactive-programming netflix rx-java

【解决方案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);
}

【讨论】：

【解决方案2】：

聚会有点晚了，但这就是 compose 存在的原因：

Observable
.from(....)
.flatMap(... -> ....)
.compose(filterAndTake(15))
.subscribe(...)

public <T> Transformer<T,T> flterAndTake(int num) {
  return source -> source
     .filter(~Filter code~)
     .take(num);
}

【讨论】：