Справочник по источникам событий в Rx

Тут важно обратить внимание, что onSuccess(T) и onComplete — взаимоисключающие. Т.е. в случае вызова первого можно не ждать второго.

Observable — наиболее часто встречающийся источник, что обусловлено его универсальностью. Он умеет как не производить события вообще, так и генерировать множество таковых, поэтому его можно использовать всегда, когда не подходят остальные варианты. Несмотря на это, у Observable есть недостаток — он совершенно не умеет обрабатывать backpressure. Для подписки на него нужны onNext(T), onError и onComplete.

Backpressure — ситуация, когда новые события поступают существенно быстрее, чем успевают обрабатываться, и начинают скапливаться в буфере, переполняя его. Это может привести к неприятностям вроде OutOfMemoryError. Подробнее можно посмотреть тут ^[1].

ConnectableObservable — прогретый вариант Observable. Все источники данных начинают выдавать свой поток событий в момент подписки. Но только не этот парень. Для этого ConnectableObservable ждёт вызова connect. Сделано это для того, чтобы несколько наблюдателей могли обозревать один поток событий, не перезапуская его при каждой подписке. Для иллюстрации приведу следующий сниппет:

val observable = Observable.fromCallable {
   Log.d("RxLogs", "observable fromCallable executed")
   Thread.sleep(1000)
}.subscribeOn(Schedulers.computation())

observable.subscribe()
observable.subscribe()
observable.subscribe()
observable.subscribe()

В консоли будет:
observable fromCallable executed
observable fromCallable executed
observable fromCallable executed
observable fromCallable executed

val connectedObservable = Observable.fromCallable {
   Log.d("RxLogs", "connectedObservable fromCallable executed")
   Thread.sleep(1000)
}.subscribeOn(Schedulers.computation())
   .publish()

connectedObservable.subscribe()
connectedObservable.subscribe()
connectedObservable.subscribe()
connectedObservable.subscribe()

connectedObservable.connect()

А в этом случае:
observable fromCallable executed

Flowable — источник, предоставляющий дополнительных операторов для обработки backpressure. Когда требуется обрабатывать более 10000 событий, происходящих быстро одно за другим, вместо Flowable рекомендуется использовать Observable.
Последний может создавать ConnectableFlowable, открывающий те же возможности, что и ConnectableObservable.

Говоря о генераторах событий, нельзя не упомянуть Subject и Processor.

Subject — класс, который может быть и источником, и обозревателем. Это позволяет использовать его, например, в разного рода контроллерах, которые будут отдавать его наружу в виде Observable и внутри оповещать как Observer. Далее пройдёмся по разным реализациям этого класса.

AsyncSubject/AsyncProcessor держит последнее событие до корректного завершения потока, после чего отдаёт его подписчикам. При возникновении ошибки никакие события проброшены не будут.

PublishSubject/PublishProcessor пробрасывает приходящие в него события дальше, пока не поступит терминальный сигнал. После конца потока или ошибки он возвращает соответствующие события.

BehaviorSubject/BehaviorProcessor работает аналогично PublishSubject/PublishProcessor, но при подписке возвращает последнее событие, если оно есть и если Subject не перешёл в терминальное состояние.

ReplaySubject/ReplayProcessor — BehaviourSubject/BehaviorProcessor на стероидах. Возвращает не одно последнее событие, а сколько душе угодно. Если подписаться на завершённый ReplaySubject или ReplayProcessor, то будут получены все накопленные данные.

Таким образом, ReplaySubject.createWithSize(1) и BehaviourSubject.create() после перехода в терминальное состояние работают по-разному. Во время подписки первый вернёт последнее событие, а второй нет. Оно же верно и для ReplayProcessor.

CompletableSubject, MaybeSubject и SingleSubject работают аналогично PublishSubject, только рассчитаны на использование с Completable, Maybe и Single соответственно.

UnicastSubject/UnicastProcessor — это фактически ReplaySubject, который следит, чтобы у него был только один подписчик. Он выбрасывает IllegalStateException при попытке повторной подписки.

Т.е. следующий сниппет

val subject = UnicastSubject.create<String>(3)
repeat(3) {
   subject.onNext(it.toString())
}

subject.onComplete()

subject.subscribe({
   Log.d("RxLogs", it)
}, {
}, {
   Log.d("RxLogs", "complete")
})

выведет в лог
0
1
2
complete

MulticastProcessor работает по аналогии с PublishProcessor, за исключением одной небольшой особенности. Он умеет обрабатывать backpressure для входящего в него потока. MulticastProcessor позволяет задать размер буфера, в котором он будет предзапрашивать элементы из upstream для будущих подписчиков.
На схеме ниже создаётся процессор с хранилищем на 2 элемента, которые он сразу запрашивает у своего источника. Поэтому, когда на него подписывается первый наблюдатель, он тут же выдаёт содержимое буфера, который моментально заполняется новыми событиями. После терминального события MulticastProcessor очищает своё хранилище и новые подписчики сразу получают завершение потока.