将Observable<Observable<Cursor>>压缩为Observable<Cursor>

Question

我有一个返回单个Cursor实例的Observable（Observable<Cursor>）。我想利用ContentObservable.fromCursor在onNext回调中获取每个游标的行。

我想到的其中一种解决方案是这样的构造：

ContentObservable.fromCursor(cursorObservable.toBlocking().first())
    .subscribe(cursor -> {
        // map to object 
        // add to outer collection
    }, e -> {}, () -> { 
        // do something with list of objects (outer collection)
    });

这看起来有点像一个 hack，因为它用到了 toBlocking().first()，但是它确实可以工作。我不喜欢它，因为大部分处理都在 onNext 回调函数中完成，我们还必须创建外部集合来保存中间结果。

我希望使用它的方式如下：

cursorObservable.map(ContentObservable::fromCursor)
    .map(fromCursor -> fromCursor.toBlocking().first())
    .map(/* map to object */)
    .toList()
    .subscribe(objects -> {
        // do something with list of objects
    }

这仍然使用了toBlocking().first()，但是不起作用，因为一旦fromCursor观察者完成了，游标就关闭了，所以没有办法将其映射到对象。有什么更好的方法将Observable<Observable<Cursor>>压平为Observable<Cursor>？

- tomrozb

Answer 1

有没有更好的方法将Observable<Observable<Cursor>>扁平化为Observable<Cursor>？

是的，您可以使用Observable.concat方法：

public static void main(String[] args) {
    final Observable<String> observable = Observable.just("1", "2");
    final Observable<Observable<String>> nested = observable.map(value -> Observable.just(value + "1", value + "2"));
    final Observable<String> flattened = Observable.concat(nested);

    flattened.subscribe(System.out::println);
}

更新

实际上有一些其他的方法可以将 Observable<Observable<Cursor>> 转换为 Observable<Cursor>:

• Observable.concat
• Observable.merge
• Observable.switchOnNext

只需选择一个对您更合适的方法即可。

更新2

另一种解决方案是稍微修改您的代码，使用 flatMap 操作符代替 map：

cursorObservable.flatMap(ContentObservable::fromCursor)
    .map(/* map to object */)
    .toList()
    .subscribe(objects -> {
        // do something with list of objects (outer collection)
    }

- Vladimir Mironov