rx-java - RxJava 2 等同于 Observable.from(Iterable).firstOrDefault？

Question

我在 RxJava 1 中有以下内容，它返回一个 Observable<SomeObject> :

// objects is a List<SomeObject>
return rx.Observable.from(objects)
                                .firstOrDefault(aDefaultObject, new Func1<SomeObject, Boolean>() {
                                    @Override
                                    public Boolean call(SomeObject aObject) {
                                        return aObject.isSomething();
                                    }
                                });

我的目标是返回一个 io.reactivex.Single<SomeObject> (RxJava 2) 代替。
我怎样才能做到这一点？

我想出了如下的东西，它是正确的吗？:

return Flowable.fromIterable(objects).filter(new Predicate<SomeObject>() {
                            @Override
                            public boolean test(SomeObject aObject) throws Exception {
                                return aObject.isSomething();
                            }
                        }).first(aDefaultObject);

Answer 1

感谢@marianosimone，正确的解决方案是:

return io.reactivex.Observable.fromIterable(objects).filter(new Predicate<SomeObject>() {
                            @Override
                            public boolean test(SomeObject aObject) throws Exception {
                                return aObject.isSomething();
                            }
                        }).first(aDefaultObject);

Flowable 也可以，但 official docs 不推荐它:
何时使用 Observable
- 最长不超过 1000 个元素的流:即，随着时间的推移，元素很少，应用程序中几乎没有机会出现 OOME。
- 您处理 GUI 事件，例如鼠标移动或触摸事件:这些事件很少会被合理地反压，而且不会那么频繁。您可以使用 Observable 处理 1000 Hz 或更低的元素频率，但无论如何都要考虑使用采样/去抖动。
- 您的流程本质上是同步的，但您的平台不支持 Java Streams，或者您错过了它的功能。使用 Observable 通常比 Flowable 具有更低的开销。 (您还可以考虑 IxJava，它针对支持 Java 6+ 的可迭代流程进行了优化)。
何时使用 Flowable
- 处理在某处以某种方式生成的 10k+ 元素，因此链可以告诉源限制它生成的数量。
- 从磁盘读取(解析)文件本质上是阻塞和基于拉取的，这在您控制背压时效果很好，例如，对于指定的请求量，您从中读取了多少行)。
- 通过 JDBC 从数据库读取也是阻塞式和基于拉式的，并且由您通过为可能的每个下游请求调用 ResultSet.next() 来控制。
- 网络(流)IO，其中网络有帮助或所用协议(protocol)支持请求一些逻辑量。
- 许多阻塞和/或基于拉取的数据源，它们最终可能在未来获得非阻塞的响应式(Reactive) API/驱动程序。