Замена EventBus на подписчиков RxJava - N, всегда прослушивающих

Я заменяю шаблон EventBus на RxJava в приложении для Android. У меня были события, чтобы предупредить любые заинтересованные стороны об обновлениях данных в моем кеше singleton. Каждый раз, когда вызывается веб-служба, данные будут обновляться, и подписчики будут предупреждены через отправленное событие.

У меня есть что-то близкое к этой настройке в RxJava с помощью AsyncSubject. Наблюдатели получают одно событие от субъекта, но затем получают событие onComplete и отписываются. Это работает как первый пользовательский интерфейс, но когда данные необходимо обновить, подписчиков не будет уведомлять. Как сообщить этим Subscribers продолжить прослушивание более onNext событий из Subject?

Мне нужен Subject, который будет сообщать о последнем элементе. PublishSubject только испускает элементы после подписки, поэтому он не совсем соответствует моим потребностям. Мои подписчики начинают наблюдать в разное время (возможно, после первого события данных), поэтому мне нужно, чтобы Subject излучал последний наблюдаемый элемент, а затем сохранил поток для последующих элементов. Кажется, что сочетание AsyncSubject и PublishSubject - это то, что мне нужно. Есть ли способ сделать это со встроенными классами, или мне нужно создать свой собственный вопрос?

WebServiceObservable OR CacheObservable
                  ^
                  |
             AsyncSubject
                  ^
                  |
                /   \
              /       \
            /           \
      UiObserver1   UiObserver2

Ответ 1

Немного поздним ответом, но немного лучшим вариантом для вашего сценария, чем BehaviorSubject, может быть BehaviorRelay из RxRelay lib. а также для более глобальных решений, когда вам нужно различное поведение, но вы хотите разделить единую точку взаимодействия между всеми модулями, которые вы можете использовать RxHub

Ответ 2

BehaviorSubject соответствует вашим потребностям. https://github.com/Netflix/RxJava/wiki/Subject#behaviorsubject

Если вам нужно более сложное поведение, вы всегда можете написать свою собственную реализацию Subject. Это кажется довольно простым.

Ответ 3

Я думаю, что это проще, если вы используете BehaviorSubject с оператором switchOnNext.

switchOnNext() преобразовать Observable, который испускает Observables (BehaviorSubject в этом примере) в один Observable, который испускает элементы, испущенные самыми последними из этих Observables

Observable, возвращаемое switchOnNext(), отписавшимся от ранее выпущенной Observable, начинает испускать элементы из последней Observable

public class PhotoModel{
  BehaviorSubject<Observable<Photo>> subject = BehaviorSubject.create(...);

  public void setUserId(String id){
   subject.onNext(Api.getUserPhoto(photoId));
  }

  public Observable<Photo> subscribeToPhoto(){
    return Observable.switchOnNext(subject);
  }
}

Когда следует использовать RxJava Observable, а когда - простой Callback на Android?