Rx Event Source Reference

RxJava is used in a large number of android applications, but at the same time many do not know other sources of events, except for Observable and maybe Flowable. They forget about the specialized classes Single, Maybe, and Completable, which are often able to add more clarity to the code.

Under the cat you will find a cheat sheet on the sources of events that exist in RxJava.

Completable is actually an Rx analog of Runnable. It is an operation that can be performed or not. If we draw an analogy with Kotlin, then this is fun completable () from the world of Rx. Accordingly, to subscribe to it, you need to implement onComplete and onError. It cannot be created from the value (Observable # just, ...) because it is not designed for this.

Single- reactive Callable, because here it is possible to return the result of the operation. Continuing the comparison with Kotlin, we can say that Single is fun single (): T {}. Thus, to subscribe to it, you must implement onSuccess (T) and onError.

Maybe - a cross between Single and Completable, because it supports one value, no values ​​and error. It is more difficult to draw an unambiguous parallel with the methods, but I think Maybe is fun maybe (): T? {}, which returns null when there is no result. It’s easy to guess that for a subscription you need to define onSuccess (T), onComplete, and onError.

It is important to note that onSuccess (T) and onComplete are mutually exclusive. Those. in case of calling the first, you can not wait for the second.

Observable is the most common source, due to its versatility. He knows how not to produce events at all, and to generate many of them, so he can be used always when other options are not suitable. Despite this, Observable has a drawback - it does not know how to handle backpressure. To subscribe to it, you need onNext (T), onError, and onComplete.

Backpressure - a situation when new events arrive much faster than they have time to process, and begin to accumulate in the buffer, overflowing it. This can lead to troubles like OutOfMemoryError. More details can be found here .

ConnectableObservable - a heated version of Observable. All data sources begin to issue their stream of events at the time of subscription. But not this guy. To do this, ConnectableObservable waits for a call to connect. This is done so that several observers can review one stream of events without restarting it at each subscription. To illustrate, I’ll give you the following snippet:

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

The console will be:
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()

And in this case: observable fromCallable executed

Flowable - a source that provides additional operators for processing backpressure. When you need to handle more than 10,000 events that occur quickly one after another, it is recommended to use it instead of Observable.

The latter can create a ConnectableFlowable, opening up the same possibilities as ConnectableObservable.

Speaking about event generators, one cannot but mention Subject and Processor.

Subject - a class that can be both a source and a browser. This allows you to use it, for example, in various kinds of controllers, which will give it outward as an Observable and inside notify as an Observer. Next, we will go through different implementations of this class.

AsyncSubject / AsyncProcessorholds the last event until the thread completes correctly, and then gives it to subscribers. If an error occurs, no events will be forwarded.



PublishSubject / PublishProcessor forwards events coming into it further until a terminal signal arrives. After the end of the stream or error, it returns the appropriate events.



BehaviorSubject / BehaviorProcessor works similarly to PublishSubject / PublishProcessor, but upon subscription it returns the last event, if any, and if Subject has not transitioned to the terminal state.



ReplaySubject / ReplayProcessor- BehaviourSubject / BehaviorProcessor on steroids. It returns not one last event, but as much as the soul desires. If you subscribe to a completed ReplaySubject or ReplayProcessor, then all the accumulated data will be received.

Thus, ReplaySubject.createWithSize (1) and BehaviourSubject.create () work differently after transitioning to the terminal state. During subscription, the first will return the last event, and the second will not. It is also true for ReplayProcessor.

CompletableSubject , MaybeSubject and SingleSubject work similarly to PublishSubject, only designed for use with Completable, Maybe and Single, respectively.

UnicastSubject / UnicastProcessor is actually a ReplaySubject that ensures that it has only one subscriber. It throws an IllegalStateException when trying to re-subscribe.



Those. next snippet

val subject = UnicastSubject.create(3)
repeat(3) {
   subject.onNext(it.toString())
}
subject.onComplete()
subject.subscribe({
   Log.d("RxLogs", it)
}, {
}, {
   Log.d("RxLogs", "complete")
})

will output to the log
0
1
2
complete

MulticastProcessor works similar to PublishProcessor, with the exception of one small feature. He knows how to handle backpressure for the incoming stream. MulticastProcessor allows you to set the size of the buffer at which it will pre-query elements from upstream for future subscribers.

In the diagram below, a processor is created with storage for 2 elements, which it immediately requests from its source. Therefore, when the first observer subscribes to it, it immediately issues the contents of the buffer, which is instantly filled with new events. After the terminal event, MulticastProcessor clears its storage and new subscribers immediately receive the completion of the stream.