Mastering Subjects in Rxjs

In this article we are going to be talking about Rxjs subjects, this is a special type of Observable that can allow us to simultaneously emit a value and subscribe it.
A Subject allows us to multicast to different observers, this is another advantage of using Subjects. A normal Observable cannot act like this because each observer maintains it's own implementation of an Observable, thus when two different observers subscribe to the Observable, they both receive different values.

import { Observable } from 'rxjs'

const observable = new Observable(subscriber => {
subscriber.next(Math.random())
})

observable.subscribe(x => console.log(x))
observable.subscribe(x => console.log(x))

The above example is the typical behavior of Observables, but subjects behave differently. Let's see somethings about Subjects.

1. Subjects are Observables. Given a Subject, we can call the Observable.next() method on it to emit values which is native to Observables. Thus we can provide an observer that will subscribe to it. Rather than create a new Observable for that subscriber, it will add the observer to it's internal list of observers.
2. Subjects are also observers and thus we can use them to consume other Observables, it has the following methods next(), error() and complete(). It will act like a a proxy, i.e multicast to other observers registered on that subject. Multicasting simply refers to a source of data that has multiple receivers. Let's create a Subject.

import { Subject } from 'rxjs'

const subject = new Subject<number>()

subject.subscribe (v => console.log(v))
subject.subscribe (v => console.log(v))

subject.next(Math.floor(Math.random() * 3))

Earlier we had discussed this; that a subject is able to emit data and later that subject will still consume the data later. The example above just showed us that. We created a subject, we subscribed to the subject, providing two separate observers. Later we used the .next() method to emit data, you will find out that the two observers got the same value. This is because Subjects multicast their values to a list of observers, the two observers we provided above were added to the Subject's list of observers and once the data was available the subject just passed out the copy of the data each observer needs from the same Observable.

import  { of, Subject } from 'rxjs'

const subject = new Subject<string>()

subject.subscribe(x => console.log('first observable', x))
subject.subscribe(x => console.log('second observable', x))

const heroes = ['thor', 'hulk', 'ironman']
const Heroes$ = of(...heroes)

Heroes$.subscribe(subject)
// first observer thor
// second observer thor
// first observer hulk
// second observer hulk
// first observer ironman
// second observer ironman

In this example we have also shown how we can use Subjects as observers, we created a subject, then created two observers for that subject. Then provided that subject as an observer to a Heroes$ observable we created with the of() operator. And we still get the multicasting too. Subjects allow us to create hot observables through multicasting. Essentially any Observable that is a multicast is a hot observable, while unicast observables are cold observables.
There are also variants of subjects, there is
BehaviorSubject, ReplaySubject, AsyncSubject.

BehaviorSubject

This is a special Observable that pushes only the current value emitted to an observer or a list of observers, although observers that are declared after a value is emitted might still get that value, however it will only get the most recent value not the entirety.

import { BehaviorSubject } from 'rxjs'

const Heroes$ = new BehaviourSubject('hawkeye')

Heroes$.subscribe(x => console.log(`first observer ${x}`))

Heroes$.next('captain America')
Heroes$.next('black widow')

Heroes$.subscribe(x => console.log(`second observer ${x}`))

Heroes$.next('deadpool')
Heroes$.next('logan')

// first observer hawkeye
// first observer captain America
// first observer black widow
// second observer black widow
// first observer deadpool
// second observer logan

We can use a behaviorSubject to hold a scoreline for a football match.

import { BehaviorSubject } from 'rxjs'

const Scoreline$ = new BehaviorSubject('0 - 0')

Scoreline$.subscribe(x => console.log(`DCU ${x} MCU`)

$Scoreline.next('1 - 0')
$Scoreline.next('1 - 1')

Scoreline$.subscribe(x => console.log(`HT DCU ${x} MCU`)

// DCU 0 - 0 MCU
// DCU 1 - 0 MCU
// DCU 1 - 1 MCU
// HT DCU 1 - 1 MCU

ReplaySubject

A ReplaySubject is quite similar to a BehaviorSubject, however a ReplaySubject will keep a record of values that has been emitted to an observer. We pass in an argument that represents how long we want the record to be, another that represents the number of milliseconds we want to store that record.

import { ReplaySubject } from 'rxjs'

const Subject$ = new ReplaySubject(3)

Subject$.subscribe(x => console.log(`observer a ${x}`))

Subject$.next(1)
Subject$.next(2)
Subject$.next(3)
Subject$.next(4)

Subject$.subscribe(x => console.log(`observer b ${x}`))

// observer a 1
// observer a 2
// observer a 3
// observer a 4
// observer b 3
// observer b 4

AsyncSubject

This is a special type of Observable that emits only its last value, after the Observable is done executing.

import { AsyncSubject } from 'rxjs'
const subject = new AsyncSubject()

subject.subscribe(x => console.log(`observer a: ${x}`))

subject.next(1)
subject.next(2)
subject.next(3)

subject.subscribe(x => console.log(`observer b: ${x}`))

subject.next(4)
subject.complete()

// observer a 4
// observer b 4

Thats all for now, we'll look more closely at Operators next, i do hope that you find this useful.