RxJS is a staple in nearly every Angular project, yet its core ideas—Observables, Observers, and Subjects—often trip up developers who are new to the framework. The mental model behind RxJS isn't always immediately obvious, and it can take a while before the library's way of thinking clicks.
To make RxJS feel more approachable, it helps to look at real-life situations that mirror how streams work. One of the simplest and most intuitive parallels is the plumbing in a home.
Think about any house or apartment: every unit has a network of pipes installed for water delivery. In RxJS terms, setting up those pipes is akin to creating a data stream with operators like from or of, which generate a flow of values.
export class WaterStreamsComponent {
stream$ = of("water")
}The stream itself—call it stream$—doesn't act on its own; it merely exists, just like the pipes installed in your walls. The water that moves through those pipes represents the data itself, whether it's a simple string, a number, an object, an array, or an event.
Consider a moment when you're thirsty and you reach for a glass. Turning the faucet releases water from the pipes; in RxJS, calling subscribe starts the flow of data from an Observable. Until you subscribe, the stream remains dormant, waiting for a listener.
export class WaterStreamsComponent implements OnInit {
stream$ = of("water")
ngOnInit() {
this.stream$.subscribe(water => console.log('drink it!', water))
}
}Leaving that faucet running indefinitely is a recipe for trouble—either you'll flood the apartment below or rack up a hefty water bill. The takeaway is clear: always close the tap when you're done. For RxJS, that means unsubscribing from a stream once it's no longer needed, which prevents memory leaks that can bog down your application over time.
export class WaterStreamsComponent implements OnInit, OnDestroy {
stream$ = of('water')
waterPipe$?: Subscription
ngOnInit() {
this.waterPipe$ = this.stream$.subscribe(
water => console.log('drink it!', water)
)
}
ngOnDestroy() {
this.waterPipe$?.unsubscribe() // <-- close water tap
}
}Shaping and Filtering the Flow
If your tap water contains impurities, you'd likely install a filter to remove metals or other particles. In RxJS, this role is played by filtering operators, with the filter operator being the go-to choice. Its behavior is reminiscent of JavaScript's array filter method. To apply it, you route the stream through the pipe function, which is where operators come together and can be chained.
import { filter } from 'rxjs/operators';
// ...
this.waterPipe$ = this.stream$.pipe(
filter((water) => water === 'water') // <-- filtering water
).subscribe(
cleanWater => console.log('drink it!', cleanWater)
)
// ...Alternatively, picture this: after a full day of Angular and RxJS work, you're looking forward to a warm shower. That comfort relies on a boiler to heat the cold water. In stream terms, heating the water is equivalent to a transformation operator like map, which converts each value as it passes through.
import { map } from 'rxjs/operators';
export class WaterStreamsComponent implements OnInit {
// ...
warmUp = (water: string) => `${water}-warm`
ngOnInit() {
this.waterPipe$ = this.stream$
.pipe(
map((water) => this.warmUp(water)) // <-- data transformation
).subscribe((warmWater) =>
console.log('take a shower!', warmWater));
}
}Operators can be layered, one after another, to build more complex pipelines. Just remember to keep a comma between each operator in the chain.
// ...
this.waterPipe$ = this.stream$
.pipe(
filter(water => water === 'water'), <-- filter first
map(water => this.warmUp(water)) <-- then warm up
).subscribe(
warmWater => console.log('take a shower!', warmWater)
)
// ...There are also times when you don't want to alter the data at all, but still need to inspect it or trigger something on the side—like how a water meter records usage without changing the flow. In RxJS, these "side-effects" are handled by the tap operator, which lets you peek at values or run procedures without modifying the stream itself.
import { tap } from 'rxjs/operators';
// ...
this.waterPipe$ = this.stream$
.pipe(
tap(
water => console.log('count water consumption', water)
)
).subscribe(
water => console.log('Drink it!', warmWater)
)
// ...When Things Go Wrong
Old plumbing can lead to leaks or even burst pipes. When that happens, acting quickly is key—often the smartest move is to shut off the water supply completely to stop the damage from spreading.
In RxJS, an error raised by any operator in the chain halts the stream's flow immediately; data stops moving downstream. The error is instead routed to error-handling operators that step in to manage the failure gracefully. Among the tools available are catchError, retry, and retryWhen.
// ...
this.waterPipe$ = this.stream$
.pipe(
map(water => { throw Error("Break the Pipe") }),
catchError(error => {
// Handle the error...
console.error('Error:', error);
// fallback value
return of('') }),
filter(water => water === 'water'), // won't execute
map(water => this.warmUp(water)) // won't execute
).subscribe(
water => console.log('No water :(', water) //<-- water is ''
)Joining Streams Together
Once you've got a handle on the pipes within your own home, it's time to zoom out. Imagine how your home's plumbing connects to the broader network—other households and the city's entire water system.
RxJS operates in a similar fashion: it's a mix of linking, merging, and drawing data from separate streams, orchestrated by transformation operators like switchMap and concatMap, as well as combination operators such as forkJoin and merge. Together, these give you the flexibility to coordinate streams in countless configurations.
export class WaterStreamsComponent implements OnInit {
stream$ = of('water')
homeWaterPipe$?: Subscription
cityWaterPipe$ = of('city water')
warmUp = (water: string) => `${water}-warm`
ngOnInit() {
this.cityWaterPipe$
.pipe(
filter((water) => water === "city water")
)
this.homeWaterPipe$ = this.stream$
.pipe(
switchMap(() => this.cityWaterPipe$), // <-- connection
map(water => this.warmUp(water)))
.subscribe(
warmWater => console.log('take a shower!', warmWater)
)
}
//...
}In the example above, purified water flows from the cityWaterPipe$ into the homeWaterPipe$. This connection is made possible by switchMap, which under the hood subscribes to the source stream, allows the data—or water—to proceed to the map operator, and ultimately delivers it to the subscriber.
Wrapping Up
I trust this walk-through of RxJS streams via water pipes has been insightful. Of course, there's no need to tear apart your actual plumbing to test out the tap operator or inspect water purity. 😃
What analogy has helped you make sense of Observables, or perhaps explain them to someone else?