java - 从 while 循环创建 Flowable

Question

我对 RxJava 还很陌生，我需要创建包含多个数据源的存储库。这对我来说很复杂，因为有几个较小的子任务我不知道如何用 RxJava 实现。

首先我有自己写的dao，它处理InputStream，并提供指定范围内的Item。目前它只是在列表中收集数据，但我想使用 flowable 来一一提供 Items；目前它提供Maybe<List<Item>> 。此外，还有一些错误需要传输到更高级别(数据源)。如EndOfFile，通知DataSource数据已完全缓存；

Dao.class :

List<Item> loadRange(int start, int number) throws ... {
    ...
    while(...) {
        ...
        //TODO contribute item to flowable
        resultList.add(new Item(...)) 

    }
    return resultList;
}

Maybe<List<Item>>刚刚创建的方法Maybe.fromCallable() ;

请帮助我!

Answer 1

类似的东西应该适用于此:

Flowable<Item> loadRange(int start, int number) {
        return Flowable.create(emitter -> {
            try {
                while (...){
                    emitter.onNext(new Item());
                }
                emitter.onComplete();
            } catch (IOException e) {
                emitter.onError(e);
            }
        }, BackpressureStrategy.BUFFER);
    }

我假设一旦循环完成，您想要完成，并向下游发送错误，而不是处理方法签名。您还可以更改 BackPressureStrategy 以适合您的用例，即 DROP、LATEST 等。

当您刚接触 RxJava 时，匿名类将是:

Flowable<Item> loadRange(int start, int number) {
        return Flowable.create(new FlowableOnSubscribe<Item>() {
            @Override public void subscribe(FlowableEmitter<Item> emitter) {
                try {
                    while (...){
                        emitter.onNext(new Item());
                    }
                    emitter.onComplete();
                } catch (IOException e) {
                    emitter.onError(e);
                }
            }
        }, BackpressureStrategy.BUFFER);
    }