我正在使用 SqlBrite/SqlDelight 在 RxJava 中实现存储库模式，用于离线数据存储和 Http 请求的改造
下面是一个示例：
protected Observable<List<Item>> getItemsFromDb() {
return database.createQuery(tableName(), selectAllStatement())
.mapToList(cursor -> selectAllMapper().map(cursor));
}
public Observable<List<Item>>getItems(){
Observable<List<Item>> server = getRequest()
.doOnNext(items -> {
BriteDatabase.Transaction transaction = database.newTransaction();
for (Item item : items){
database.insert(tableName(), contentValues(item));
}
transaction.markSuccessful();
transaction.end();
})
.flatMap(items -> getItemsFromDbById())
.delaySubscription(200, TimeUnit.MILLISECONDS);
Observable<List<Item>> db = getItemsFromDbById(id)
.filter(items -> items != null && items.size() > 0);
return Observable.amb(db, server).doOnSubscribe(() -> server.subscribe(items -> {}, throwable -> {}));
}
当前的实现使用Observable.amb获取 2 个流中的最新流db并在db有数据或服务器的情况下返回流。为了防止在没有互联网的情况下及早失败，server有一个带有200msdelaySubscription。
我尝试使用Observable.concat但 SqlBrite 流从不调用onComplete因此永远不会触发server可观察性。
我也尝试了Observable.combineLatest不起作用，因为它在发出任何东西之前一直等待server可观察的返回数据，Observable.switchOnNext也不起作用。
我正在寻找的是一个存储库，它：
- 保持对 SqlBrite (DB) 的订阅打开，以防数据库更新
- 始终从服务器获取数据并将其写入数据库
- 如果数据库中没有任何内容并且网络请求仍在进行，则不应发出空结果。这是因为用户在第一次加载的情况下应该看到一个进度条。
这就是解决上述问题的方法，即从 2 个源(本地和远程)获取数据并仅在需要时向 UI 发送更新。
数据类包装数据并存储数据源
class Data<T> {
static final int STATE_LOCAL = 0;
static final int STATE_SERVER = 1;
private T data;
private int state;
Data(T data, int state) {
this.data = data;
this.state = state;
}
public int getState() { return state; }
public T getData() { return data; }
}
。
public Observable<Model> getData(long id) {
// Used to cache data and compare it with server data, so we can avoid unnecessary UI updates
Subject<Data<Model>> publishSubject = BehaviorSubject.create();
publishSubject.onNext(new Data<>(null, Data.STATE_LOCAL));
Observable<Data<Model>> server = getRequest()
.map(items -> new Data<>(items, Data.STATE_SERVER))
// Here we are combining data from server and our `BehaviorSubject`
// If any one has ideas how to do this without the subject, I'll be glad to hear it.
.flatMap(items -> Observable.zip(publishSubject.take(1), Observable.just(items), Pair::new))
.flatMap(oldNewPair -> {
// Here we are comparing old and new data to see if there was any new data returned from server
Data<Model> prevData = oldNewPair.first;
Data<Model> newData = oldNewPair.second;
//Could be any condition to compare the old and new data
if (prevData.data != null && prevData.data.updated_at() == newData.data.updated_at())
return Observable.just(prevData);
else
return database.insert(tableName(), contentValues(newData));
return getFromDb(id)
.map(item -> new Data<>(item, Data.STATE_LOCAL))
.onErrorResumeNext(server)
.doOnNext(item -> {
publishSubject.onNext(item);
if (item.getState() == Data.STATE_LOCAL)
server.subscribeOn(Schedulers.io()).observeOn(Schedulers.io()).subscribe();
})
.map(item -> item.data);
}
此解决方案不使用amb，并使用 BehaviorSubject 来解决以下问题：
不使用
delaySubscription(早期用于防止在没有互联网的情况下早期失败。早些时候，每次对服务器进行两次调用时，在这种情况下都可以解决。
你的代码直接与你想要做的事情相矛盾。这一行：
Observable<List<Item>> db = getItemsFromDbById(id)
.filter(items -> items != null && items.size() > 0);
与自身相矛盾，因为您返回单个数据库查询的项目并将其命名为 db - 就好像数据库(或其引用)本身一样。从这一点来看，很明显您提供的代码无法提供帮助。
有许多存储库模式的 Java 模板可用。例如： https://www.bignerdranch.com/blog/the-rxjava-repository-pattern/
如果这还不够有帮助，请尝试提供至少与您描述的内容相距甚远的代码。