RxSwift course - Subjects

Last time we talked about observable and how to subscribe for values. This time I'll show You how to make sequences more "dynamic"

Subject

Subject in Rx works like an observer, you can send value to it and the subject will pass it down. RxSwift provides three basic subjects:

PublishSubject - most basic one, is empty when created and then emits only new elements to observers, in other terms, this one is hot.
BehaviourSubject - this one has state, it starts with the initial value and always emits the last value to its observer when subscribed. Useful with UI stuff when you want to have an initial state for controls.
ReplaySubject - like the behavior subject, but you can specify the size of a buffer and the whole buffer will be replayed when subscribed.

For the test environment, we will use the same starter project as before, found here.

Publish subject

Open starter project and run code below:

let disposeBag = DisposeBag()

let publishSubject = PublishSubject<String>()

publishSubject.onNext("Lorem")

publishSubject
    .subscribe(onNext: { print("first subscription \($0)") })
    .disposed(by: disposeBag)

let toBeDisposed = publishSubject
    .subscribe(onNext: { print("disposable subscription \($0)") })

publishSubject.onNext("Ipsum")

publishSubject
    .subscribe(onNext: { print("second subscription \($0)") })
    .disposed(by: disposeBag)

publishSubject.onNext("dolor")

publishSubject
    .subscribe(onNext: { print("third subscription \($0)") })
    .disposed(by: disposeBag)

publishSubject.onNext("sit")

toBeDisposed.dispose()

publishSubject.onNext("amet")

I'm creating a couple of subscriptions for our subject, one of them will be disposed before the program ends. The output will tell us what is going on:)

first subscription Ipsum
disposable subscription Ipsum
first subscription dolor
disposable subscription dolor
second subscription dolor
first subscription sit
disposable subscription sit
second subscription sit
third subscription sit
first subscription amet
second subscription amet
third subscription amet
Program ended with exit code: 0

As we can see, publish subject is a hot observable, it doesn't share any state and if a value is sent before subscription it will not be replayed. Also, disposing one observable doesn't kill all observables, just one disposed - which is logical, but worth mentioning at this point.

Behavior Subject

I'll use the same code, but change the type of subject to behavior to see the difference:

let disposeBag = DisposeBag()

let behaviourSubject = BehaviorSubject<String>(value: "Lorem")

behaviourSubject
    .subscribe(onNext: { print("first subscription \($0)") })
    .disposed(by: disposeBag)

let toBeDisposed = behaviourSubject
    .subscribe(onNext: { print("disposable subscription \($0)") })

behaviourSubject.onNext("Ipsum")

behaviourSubject
    .subscribe(onNext: { print("second subscription \($0)") })
    .disposed(by: disposeBag)

behaviourSubject.onNext("dolor")

behaviourSubject
    .subscribe(onNext: { print("third subscription \($0)") })
    .disposed(by: disposeBag)

behaviourSubject.onNext("sit")

toBeDisposed.dispose()

behaviourSubject.onNext("amet")

And the output:

first subscription Lorem
disposable subscription Lorem
first subscription Ipsum
disposable subscription Ipsum
second subscription Ipsum
first subscription dolor
disposable subscription dolor
second subscription dolor
third subscription dolor
first subscription sit
disposable subscription sit
second subscription sit
third subscription sit
first subscription amet
second subscription amet
third subscription amet
Program ended with exit code: 0

BehaviourSubject replays last value to new subscribers, for a first subscription it is the initial value, for rest last sent value. The third subscription is immediately printing "dolor" upon subscription, as expected.

Replay Subject

In a similar case, I'll change subject to the Replay one and see what is printed to console:

let disposeBag = DisposeBag()

let replaySubject = ReplaySubject<String>.create(bufferSize: 3)

replaySubject.onNext("Latin")
replaySubject.onNext("Text:")

replaySubject
    .subscribe(onNext: { print("first subscription \($0)") })
    .disposed(by: disposeBag)

let toBeDisposed = replaySubject
    .subscribe(onNext: { print("disposable subscription \($0)") })

replaySubject.onNext("Ipsum")

replaySubject
    .subscribe(onNext: { print("second subscription \($0)") })
    .disposed(by: disposeBag)

replaySubject.onNext("dolor")

replaySubject
    .subscribe(onNext: { print("third subscription \($0)") })
    .disposed(by: disposeBag)

replaySubject.onNext("sit")

toBeDisposed.dispose()

replaySubject.onNext("amet")

And output:

first subscription Latin
first subscription Text:
disposable subscription Latin
disposable subscription Text:
first subscription Ipsum
disposable subscription Ipsum
second subscription Latin
second subscription Text:
second subscription Ipsum
first subscription dolor
disposable subscription dolor
second subscription dolor
third subscription Text:
third subscription Ipsum
third subscription dolor
first subscription sit
disposable subscription sit
second subscription sit
third subscription sit
first subscription amet
second subscription amet
third subscription amet
Program ended with exit code: 0

ReplaySubject will emit up to buffer size elements when subscribed. The first subscription is printing two values, that are sent before the subscription. The third subscription is added when the buffer is already full and it is replaying the last three values, as expected. At this point, the first value is discarded, since the buffer holds only three values.