我是 RxJava 的新手，我需要創建包含多個數據源的存儲庫。這對我來說很復雜，因為有幾個較小的子任務我不知道如何用 RxJava 實現。先是自己寫了dao，處理InputStream，提供指定范圍內的Item。目前它只是在列表中收集數據，但我想使用 flowable 一個一個地提供項目；目前它提供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()；
1 回答
千巷貓影
TA貢獻1829條經驗 獲得超7個贊
這樣的事情應該適用于此：
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);
}
添加回答
舉報
0/150
提交
取消