RxJS Primer

Most state in the Browser SDK — devices, participants, capabilities, call status — is exposed as RxJS observables. This page covers the subset of RxJS used by the SDK. For everything else, see the RxJS docs.

Observables and subscriptions

An observable is a stream of values over time. It does nothing until .subscribe() is called; .subscribe() returns a subscription, and .unsubscribe() ends it. Subscriptions left open are the main source of memory leaks.

SDK observables behave like BehaviorSubjects: subscribing emits the current value synchronously, then every subsequent change.

Transform a stream with .pipe() and operators (filter, map, take, combineLatest, switchMap, debounceTime, etc.).

The `$` suffix

A property ending in $ is the observable. The same name without $ is the current snapshot.

1 const status = call.status;        // current value
2 call.status$.subscribe(handler);   // current value + every change

Snapshots are not always populated. For state the SDK already holds in memory (call.status, call.participants, device lists), the snapshot is the current value and reading it is fine. For lazily-loaded collections — directory.addresses, address.textMessages, address.history — the snapshot starts empty until something subscribes to the corresponding $ observable (and, for paginated collections, until loadMore() is called). Reading the snapshot synchronously right after connecting will give you []. Subscribe to the $ form, then call loadMore() to trigger the first page.

Subscribing

1 const sub = client.audioInputDevices$.subscribe((devices) => {
2   console.log("mics:", devices);
3 });
4 
5 sub.unsubscribe();

The first emission fires synchronously with the current device list; subsequent emissions fire on device changes.

Common patterns

Wait for a value, then proceed once. take(1) ends the subscription after the first match.

1 import { filter, take } from "rxjs";
2 
3 client.ready$
4   .pipe(filter(Boolean), take(1))
5   .subscribe(async () => {
6     const call = await client.dial(destination);
7   });

React on every match.

1 import { filter } from "rxjs";
2 
3 call.status$.pipe(filter((s) => s === "connected")).subscribe(showCallControls);

Combine the latest of multiple streams.

1 import { combineLatest } from "rxjs";
2 
3 combineLatest([client.audioInputDevices$, client.selectedAudioInputDevice$])
4   .subscribe(([devices, selected]) => {
5     const activeMic = devices.find((d) => d.deviceId === selected?.deviceId);
6   });

Skip the initial value when only changes matter.

1 import { skip } from "rxjs";
2 
3 call.status$.pipe(skip(1)).subscribe(showStatusNotification);

Throttle high-frequency streams.

1 import { debounceTime } from "rxjs";
2 
3 call.localAudioLevel$.pipe(debounceTime(100)).subscribe(updateVolumeIndicator);

Cleanup

Two patterns:

1 class CallManager {
2   subs = [];
3   start(call) {
4     this.subs.push(
5       call.status$.subscribe(this.onStatus),
6       call.participants$.subscribe(this.onParticipants),
7     );
8   }
9   stop() {
10     this.subs.forEach((s) => s.unsubscribe());
11     this.subs = [];
12   }
13 }

1 import { Subject, takeUntil } from "rxjs";
2 
3 const destroy$ = new Subject();
4 
5 call.status$.pipe(takeUntil(destroy$)).subscribe(updateStatus);
6 call.participants$.pipe(takeUntil(destroy$)).subscribe(updateParticipants);
7 
8 destroy$.next();
9 destroy$.complete();

takeUntil scales better with many subscriptions; an array is fine for a few.

References

RxJS docs
Observables
Operators
Subjects (including BehaviorSubject)

Observables and subscriptions

SDK observables behave like BehaviorSubjects: subscribing emits the current value synchronously, then every subsequent change.

Transform a stream with .pipe() and operators (filter, map, take, combineLatest, switchMap, debounceTime, etc.).

The `$` suffix

A property ending in $ is the observable. The same name without $ is the current snapshot.

1 const status = call.status;        // current value
2 call.status$.subscribe(handler);   // current value + every change

Subscribing

1 const sub = client.audioInputDevices$.subscribe((devices) => {
2   console.log("mics:", devices);
3 });
4 
5 sub.unsubscribe();

The first emission fires synchronously with the current device list; subsequent emissions fire on device changes.

Common patterns

Wait for a value, then proceed once. take(1) ends the subscription after the first match.

1 import { filter, take } from "rxjs";
2 
3 client.ready$
4   .pipe(filter(Boolean), take(1))
5   .subscribe(async () => {
6     const call = await client.dial(destination);
7   });

React on every match.

1 import { filter } from "rxjs";
2 
3 call.status$.pipe(filter((s) => s === "connected")).subscribe(showCallControls);

Combine the latest of multiple streams.

1 import { combineLatest } from "rxjs";
2 
3 combineLatest([client.audioInputDevices$, client.selectedAudioInputDevice$])
4   .subscribe(([devices, selected]) => {
5     const activeMic = devices.find((d) => d.deviceId === selected?.deviceId);
6   });

Skip the initial value when only changes matter.

1 import { skip } from "rxjs";
2 
3 call.status$.pipe(skip(1)).subscribe(showStatusNotification);

Throttle high-frequency streams.

1 import { debounceTime } from "rxjs";
2 
3 call.localAudioLevel$.pipe(debounceTime(100)).subscribe(updateVolumeIndicator);

Cleanup

Two patterns:

1 class CallManager {
2   subs = [];
3   start(call) {
4     this.subs.push(
5       call.status$.subscribe(this.onStatus),
6       call.participants$.subscribe(this.onParticipants),
7     );
8   }
9   stop() {
10     this.subs.forEach((s) => s.unsubscribe());
11     this.subs = [];
12   }
13 }

1 import { Subject, takeUntil } from "rxjs";
2 
3 const destroy$ = new Subject();
4 
5 call.status$.pipe(takeUntil(destroy$)).subscribe(updateStatus);
6 call.participants$.pipe(takeUntil(destroy$)).subscribe(updateParticipants);
7 
8 destroy$.next();
9 destroy$.complete();

takeUntil scales better with many subscriptions; an array is fine for a few.

References

RxJS docs
Observables
Operators
Subjects (including BehaviorSubject)

1	const status = call.status; // current value
2	call.status$.subscribe(handler); // current value + every change

1	const sub = client.audioInputDevices$.subscribe((devices) => {
2	console.log("mics:", devices);
3	});
4
5	sub.unsubscribe();

1	import { filter, take } from "rxjs";
2
3	client.ready$
4	.pipe(filter(Boolean), take(1))
5	.subscribe(async () => {
6	const call = await client.dial(destination);
7	});

1	import { filter } from "rxjs";
2
3	call.status$.pipe(filter((s) => s === "connected")).subscribe(showCallControls);

1	import { combineLatest } from "rxjs";
2
3	combineLatest([client.audioInputDevices$, client.selectedAudioInputDevice$])
4	.subscribe(([devices, selected]) => {
5	const activeMic = devices.find((d) => d.deviceId === selected?.deviceId);
6	});

1	import { skip } from "rxjs";
2
3	call.status$.pipe(skip(1)).subscribe(showStatusNotification);

1	import { debounceTime } from "rxjs";
2
3	call.localAudioLevel$.pipe(debounceTime(100)).subscribe(updateVolumeIndicator);

1	class CallManager {
2	subs = [];
3	start(call) {
4	this.subs.push(
5	call.status$.subscribe(this.onStatus),
6	call.participants$.subscribe(this.onParticipants),
7	);
8	}
9	stop() {
10	this.subs.forEach((s) => s.unsubscribe());
11	this.subs = [];
12	}
13	}

1	import { Subject, takeUntil } from "rxjs";
2
3	const destroy$ = new Subject();
4
5	call.status$.pipe(takeUntil(destroy$)).subscribe(updateStatus);
6	call.participants$.pipe(takeUntil(destroy$)).subscribe(updateParticipants);
7
8	destroy$.next();
9	destroy$.complete();

Observables and subscriptions

The $ suffix

Subscribing

Common patterns

Cleanup

References

Observables and subscriptions

The $ suffix

Subscribing

Common patterns

Cleanup

References

The `$` suffix

The `$` suffix