Шаблон репозитория с SqlBrite/SqlDelight (автономная база данных) и дооснащение (запрос Http)

Я реализую шаблон репозитория в RxJava, используя SqlBrite/SqlDelight для автономного хранения данных и модификации для запросов 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 имеет delaySubscription на нем с 200ms.

Я попытался использовать Observable.concat, но поток SqlBrite никогда не вызывает onComplete, поэтому server наблюдаемый никогда не запускается.

Я также пробовал Observable.combineLatest, который не работал, потому что он продолжает ждать, пока server наблюдается, чтобы возвращать данные, прежде чем что-либо исправить, и Observable.switchOnNext тоже не работает.

Я ищу репозиторий, который:

Открывает подписку на SqlBrite (DB) в случае обновлений БД
Всегда извлекает данные с сервера и записывает их в базу данных
Не следует выводить пустой результат в случае, если в базе данных ничего не было, и сетевой запрос все еще продолжается. Это, потому что пользователь должен увидеть индикатор выполнения в случае первой загрузки.

Ответ 1

Вот как вы можете решить проблему выше, то есть получать данные из двух источников (локальных и удаленных) и отправлять обновление в интерфейс только тогда, когда это необходимо.

Класс данных обертывает ваши данные и также сохраняет источник данных

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 (ранее использовалось для предотвращения раннего отказа в случае отсутствия Интернета.)
Ранее, каждый раз, когда на сервер решались два вызова, которые решаются в этом случае.

Ответ 2

Вы неправильно кодируете то, что хотите. Эта строка:

     Observable<List<Item>> db = getItemsFromDbById(id)
             .filter(items -> items != null && items.size() > 0);

Является противоречием с самим собой, потому что вы возвращаете отдельные элементы запроса базы данных и называете его db - как если бы база данных (или ее ссылка) сама. С этого момента ясно, что предоставленный вами код не может быть помог.

В шаблоне репозитория доступно много шаблонов java. Например: https://www.bignerdranch.com/blog/the-rxjava-repository-pattern/

Если это не помогло, попробуйте предоставить код, который, по крайней мере, отдаленно описывает то, что вы описываете.