RxAndroid filter Observable<List<Item>>

Question

我使用RxAndroid，我想过滤Observable<List<Item>>的结果。

代码如下:

Observable<List<Item>> observable = 
Observable.create((Observable.OnSubscribe<List<Item>>) subscriber -> {
        subscriber.onNext(ItemManager.getItems());
        subscriber.onCompleted();
    });

我想使用.filter()只获得有效的项目。这样做将是完美的:

observable.filter(new Func1<Item, Boolean>() {
    @Override
        public Boolean call(Item item) {
            return item.isValid();
    }
});

然而，.filter()强迫我实现新的Func1():

observable.filter(new Func1<List<Item>, Boolean>() {
        @Override
            public Boolean call(Item item) {
        }
    });

有人知道如何解决这个问题吗?

提前感谢:)

Answer 1

     Observable.defer(() -> Observable.just(ItemManager.getItems())
            .subscribeOn(Schedulers.io())
            .flatMapIterable(items -> items)
            .filter(item -> item.isValid())
            .toList();

Answer 2

用from函数创建Observable:

    Observable<Item> observable = Observable.from(ItemManager.getItems())
    .filter(new Func1<Item, Boolean>() {
        @Override
        public Boolean call(Item item) {
            return item.isValid();
        }
    });
    observable.subscribe(System.out::println);

:

由于您需要异步调用ItemManager.getItems()，您需要执行以下操作:

Observable<List<Item>> observable = 
Observable.create((Observable.OnSubscribe<Item>) subscriber -> {
        for (Item item : ItemManager.getItems()) {
            subscriber.onNext(item);
        }
        subscriber.onCompleted();
    })
.filter(item -> item.isValid())
.subscribeOn(Schedulers.io());

这样订阅和对ItemManager.getItems()的调用将在io线程上执行。

Answer 3

我找到了这个解决方案，但我不知道这是不是一个好方法:

Observable<List<Item>> observable = 
Observable.create((Observable.OnSubscribe<Item>) subscriber -> {
        for (Item item : ItemManager.getItems()) {
            subscriber.onNext(item);
        }
        subscriber.onCompleted();
    })
    .filter(item -> item.isValid());