我是 RxJava 的新手，我需要創(chuàng)建包含多個數(shù)據(jù)源的存儲庫。這對我來說很復(fù)雜，因為有幾個較小的子任務(wù)我不知道如何用 RxJava 實現(xiàn)。先是自己寫了dao，處理InputStream，提供指定范圍內(nèi)的Item。目前它只是在列表中收集數(shù)據(jù)，但我想使用 flowable 一個一個地提供項目；目前它提供Maybe<List<Item>>。還有一些錯誤需要傳輸?shù)礁呒墑e（數(shù)據(jù)源）。比如EndOfFile，通知DataSource數(shù)據(jù)緩存完畢；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>>剛剛創(chuàng)建的方法Maybe.fromCallable()；
1 回答
千巷貓影
TA貢獻1829條經(jīng)驗 獲得超7個贊
這樣的事情應(yīng)該適用于此：
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);
}
我假設(shè)一旦循環(huán)完成你想要完成，也向下游發(fā)送錯誤，而不是處理方法簽名。您也可以更改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
提交
取消