Правильный поток в RxJava с дооснащением и недвижимостью

Question

Я реализую сетевой API с комбинацией RxJava и Retrofit, и я использую Realm в качестве моей базы данных. Я получил это довольно много работы, но мне интересно, если это правильный подход и поток событий. Итак, вот RetrofitApiManager.

public class RetrofitApiManager { 

    private static final String BASE_URL = "***"; 

    private final ShopApi shopApi; 

    public RetrofitApiManager(OkHttpClient okHttpClient) { 

     // GSON INITIALIZATION 

     Retrofit retrofit = new Retrofit.Builder() 
       .client(okHttpClient) 
       .addCallAdapterFactory(RxJavaCallAdapterFactory.create()) 
       .addConverterFactory(GsonConverterFactory.create(gson)) 
       .baseUrl(BASE_URL) 
       .build(); 

     shopApi = retrofit.create(ShopApi.class); 
    } 

    public Observable<RealmResults<Shop>> getShops() { 
     return shopApi.getShops() 
       .subscribeOn(Schedulers.io()) 
       .observeOn(AndroidSchedulers.mainThread()) 
       .doOnNext(response -> { 
        Realm realm = Realm.getDefaultInstance(); 
        realm.executeTransaction(realm1 -> 
          realm1.copyToRealmOrUpdate(response.shops)); 
        realm.close(); 
       }) 
       .flatMap(response -> { 
        Realm realm = Realm.getDefaultInstance(); 
        Observable<RealmResults<Shop>> results = realm.where(Shop.class) 
          .findAllAsync() 
          .asObservable() 
          .filter(RealmResults::isLoaded); 
        realm.close(); 
        return results; 
       }); 
    } 
} 

А вот вызов, чтобы получить RealmResults<Shop> внутри Fragment.

realm.where(Shop.class) 
     .findAllAsync() 
     .asObservable() 
     .filter(RealmResults::isLoaded) 
     .first() 
     .flatMap(shops -> 
       shops.isEmpty() ? retrofitApiManager.getShops() : Observable.just(shops)) 
     .subscribe(
       shops -> initRecyclerView(), 
       throwable -> processError(throwable)); 

Вот мои вопросы:

1. ли это правильный подход к цепи событий, как в примере выше, или я должен управлять им по-другому?

2. Можно ли использовать Realm экземпляр в методе getShops() и закрыть его там, или было бы лучше передать его в качестве аргумента, а затем каким-то образом управлять им? Хотя эта идея кажется немного проблематичной для потоков и вызывает Realm.close() всегда в нужное время.

Answer 1

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

2)

public Observable<RealmResults<Shop>> getShops() { 
     return shopApi.getShops() 
       .subscribeOn(Schedulers.io()) 
       .observeOn(AndroidSchedulers.mainThread()) 
       .doOnNext(response -> { 
        try(Realm realm = Realm.getDefaultInstance()) { 
         realm.executeTransaction(realm1 -> 
          realm1.insertOrUpdate(response.shops)); 
        } // auto-close 
       }) 
       .flatMap(response -> { 
        try(Realm realm = Realm.getDefaultInstance()) { 
         Observable<RealmResults<Shop>> results = realm.where(Shop.class) 
          .findAllAsync() 
          .asObservable() 
          .filter(RealmResults::isLoaded); 
        } // auto-close 
        return results; 
       }); 
    } 

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

Если вы хотите, вы можете использовать copyFromRealm(), чтобы получить неуправляемые данные, которые можно перемещать по потокам и больше не подключаться к Realm, но они также потеряют свои функции обновления и увеличат объем памяти.

Это, вероятно, сделать это вместо того, чтобы:

public Observable<RealmResults<Shop>> getShops() { 
     return shopApi.getShops() 
       .subscribeOn(Schedulers.io()) 
       .doOnNext(response -> { 
        try(Realm realm = Realm.getDefaultInstance()) { 
         realm.executeTransaction(realm1 -> 
          realm1.copyToRealmOrUpdate(response.shops)); 
        } // auto-close 
       }) 
       .observeOn(AndroidSchedulers.mainThread()) 
       .flatMap(response -> { 
        Observable<RealmResults<Shop>> results = realm.where(Shop.class) 
          .findAllAsync() 
          .asObservable() 
          .filter(RealmResults::isLoaded); 
        return results; 
       }); 

В качестве альтернативы вы можете обращаться с сетевой запрос, как побочный эффект, и только зависит от Realm уведомления вас, когда есть изменения (лучше подход ИМО, как вы отделить сеть от доступ к БД, которая является, например, то, что Repository картина о)

public Observable<RealmResults<Shop>> getShops() { 
    // Realm will automatically notify this observable whenever data is saved from the network 
    return realm.where(Shop.class).findAllAsync().asObservable() 
      .filter(RealmResults::isLoaded) 
      .doOnNext(results -> { 
       if (results.size() == 0) { 
        loadShopsFromNetwork(); 
       } 
      }); 
} 

private void loadShopsFromNetwork() { 
    shopApi.getShops() 
      .subscribeOn(Schedulers.io()) 
      .subscribe(response -> { 
       try(Realm realm = Realm.getDefaultInstance()) { 
        realm.executeTransaction(r -> r.insertOrUpdate(response.shops)); 
       } // auto-close 
      }); 
} 

Answer 2

Что Кристиан Мельхиор упомянул в своем ответе, имеет смысл, и должно решить проблему вы имеете в руке, но вниз линии такой подход может ввести другие вопросы).

В хорошей архитектуре все основные модули (или библиотеки) должны быть изолированы от остальной части кода. Поскольку Realm, RealmObject или RealmResult не могут быть переданы по потокам, еще важнее сделать Realm & Операции, связанные с риском, выделенные из остальной части кода.

Для каждого вашего класса jsonModel у вас должен быть класс realmModel и DAO (объект доступа к данным). Идея здесь заключается в том, что кроме класса DAO ни один из классов не должен знать или получать доступ к realmModel или Realm. Класс DAO принимает jsonModel, преобразует в realmModel, выполняет операции чтения/записи/редактирования/удаления, для операций чтения DAO преобразует realmModel в jsonModel и возвращается с ним.

Таким образом, легко поддерживать Realm, избегать всех проблем, связанных с Thread, легко тестировать и отлаживать.

Вот статья о Realm передовой практике с хорошей Architechture https://medium.com/@Viraj.Tank/realm-integration-in-android-best-practices-449919d25f2f

Также образец проекта, демонстрирующего Интеграция Realm на Android с MVP (Model View Presenter), RxJava, дооборудования, Dagger, аннотаций & тестирования. https://github.com/viraj49/Realm_android-injection-rx-test

Answer 3

В моем случае, я, кажется, определили запрос для RealmRecyclerViewAdapter как это:

recyclerView.setAdapter(new CatAdapter(getContext(), 
      realm.where(Cat.class).findAllSortedAsync(CatFields.RANK, Sort.ASCENDING))); 

И в противном случае определено условие для Модернизированный с RxJava скачать больше вещей, когда условие:

Subscription downloadCats = Observable.create(new RecyclerViewScrollBottomOnSubscribe(recyclerView)) 
      .filter(isScrollEvent -> isScrollEvent || realm.where(Cat.class).count() <= 0) 
      .switchMap(isScrollEvent -> catService.getCats().subscribeOn(Schedulers.io())) // RETROFIT 
      .retry() 
      .subscribe(catsBO -> { 
       try(Realm outRealm = Realm.getDefaultInstance()) { 
        outRealm.executeTransaction((realm) -> { 
         Cat defaultCat = new Cat(); 
         long rank; 
         if(realm.where(Cat.class).count() > 0) { 
          rank = realm.where(Cat.class).max(Cat.Fields.RANK.getField()).longValue(); 
         } else { 
          rank = 0; 
         } 
         for(CatBO catBO : catsBO.getCats()) { 
          defaultCat.setId(catBO.getId()); 
          defaultCat.setRank(++rank); 
          defaultCat.setSourceUrl(catBO.getSourceUrl()); 
          defaultCat.setUrl(catBO.getUrl()); 
          realm.insertOrUpdate(defaultCat); 
         } 
        }); 
       } 
      }, throwable -> { 
       Log.e(TAG, "An error occurred", throwable); 
      }); 

И это, например, поиск на основе ввода правку текста:

Subscription filterDogs = RxTextView.textChanges(editText) 
        .switchMap((charSequence) -> 
          realm.where(Dog.class) 
           .contains(DogFields.NAME, charSequence.toString()) 
           .findAllAsyncSorted(DogFields.NAME, Sort.ASCENDING) 
           .asObservable()) 
        .filter(RealmResults::isLoaded) 
        .subscribe(dogs -> realmRecyclerAdapter.updateData(dogs));