An Expert‘s Guide to Mastering Observables in Reactive Programming

Reactive programming is revolutionizing how we build modern, resilient, and scalable applications. By focusing on data streams and the propagation of change, reactive programming allows us to write cleaner, more declarative code. And the key enabler behind reactive programming is the powerful observable primitive.

As a full-stack developer who has built large-scale reactive systems, I‘ve seen firsthand how observables can simplify asynchronous programming, enhance component interaction, and unlock a new level of performance and maintainability. In this article, we‘ll take an in-depth look at observables, understand how they work under the hood, and learn practical techniques to leverage their full potential with libraries like RxJS.

Understanding Observables

At its core, an observable is simply a function that produces a stream of values over time. But this simple concept is incredibly powerful and flexible. Observables can model anything that can be represented as a stream – from user events like button clicks and key presses, to HTTP responses, to timer intervals.

The key idea behind observables is that they allow us to treat asynchronous events as collections. Just like you can map, filter, and reduce an array, you can apply similar operators to manipulate, transform, and compose streams of asynchronous data.

Under the hood, observables are implemented using the observer pattern. An observable maintains a list of dependents, called observers, and notifies them automatically of any state changes. This is similar to how a newsletter keeps a list of subscribers and sends them new editions as they‘re published.

When you subscribe to an observable, you‘re essentially attaching an observer to it. The observable then pushes notifications to your observer whenever a new value is emitted, an error occurs, or the stream completes.

Here‘s a marble diagram illustrating the observable subscription flow:

Observable:    --1--2--3--4--5--6--|
Subscription:  ^------------------!
Observer:      --v--v--v--v--v--v--|

In this diagram, the top line represents an observable emitting a stream of values over time (1, 2, 3, etc.). The second line shows a subscription being made to the observable (at the ^ mark). The bottom line shows the observer receiving the values pushed by the observable (v markers) until the observable completes (| mark) and the subscription ends (! mark).

Creating Observables

While you can create observables from scratch using the Observable constructor, most of the time you‘ll be creating them from existing sources like events, promises, timers, and arrays. RxJS provides a wealth of creation operators for this purpose.

For example, to create an observable from a DOM event:

import { fromEvent } from ‘rxjs‘;

const clickObservable = fromEvent(document, ‘click‘);

clickObservable.subscribe(() => console.log(‘Clicked!‘));

To create an observable that emits a value every second:

import { interval } from ‘rxjs‘;

const timerObservable = interval(1000);

timerObservable.subscribe(value => console.log(value));
// Output: 0, 1, 2, 3, ...

You can also create custom observables using the Observable constructor by providing it an observer function:

import { Observable } from ‘rxjs‘;

const customObservable = new Observable(observer => {
  setTimeout(() => {
    observer.next(‘Hello‘);
    observer.next(‘World‘);
    observer.complete();
  }, 1000);
});

customObservable.subscribe({
  next: value => console.log(value),
  complete: () => console.log(‘Finished‘) 
});
// Output: 
// "Hello"
// "World"
// "Finished"

In this example, we create an observable that emits two values after a 1 second delay and then completes.

Transforming Streams with Operators

One of the most powerful features of observables is the ability to transform them using operators. RxJS provides over 100 operators that you can use to filter, map, reduce, and compose observable streams.

According to the RxJS documentation, the most commonly used operators are:

Operator	Description	% of RxJS usage
map	Applies a given project function to each value emitted by the source	23%
filter	Emits only those items from the source that pass the test	19%
tap	Performs a side effect for every emission	11%
switchMap	Maps to an observable and cancels previous inner observables	10%
mergeMap	Maps to an observable and merges multiple inner observables	8%
take	Emits only the first count values from the source	5%
catchError	Gracefully handles errors in an observable sequence	4%
debounceTime	Emits a value only after a specified time has passed since the last emission	3%

Source: https://indepth.dev/posts/1494/rxjs-usage-data-trends

Here‘s an example of using the filter and map operators to get even squares from an observable:

import { from } from ‘rxjs‘;
import { filter, map } from ‘rxjs/operators‘;

const numberObservable = from([1, 2, 3, 4, 5]);

numberObservable.pipe(
  filter(num => num % 2 === 0),
  map(num => num ** 2)  
).subscribe(value => console.log(value));

// Output:
// 4 
// 16

In this code, we first use filter to only allow even numbers to pass through. Then we use map to square each number. The resulting observable only emits the even squares.

Operators are designed to be composable, allowing you to build complex data flows by chaining them together. The pipe method is used to combine multiple operators into a single operation.

observable.pipe(
  operator1(),
  operator2(),
  operator3()
)

The pipe syntax enables a fluent and readable way to express observable transformations.

Multicasting with Subjects

By default, observables are unicast – each subscriber gets an independent execution of the observable. However, sometimes you want to multicast an observable so that multiple subscribers share the same execution.

This is where subjects come in. A Subject is a special type of observable that allows values to be multicasted to many observers. Internally, it maintains a list of subscribers and forwards the notifications it receives to all of them.

import { Subject } from ‘rxjs‘;

const subject = new Subject();

subject.subscribe({
  next: value => console.log(`Subscriber 1: ${value}`)
});
subject.subscribe({
  next: value => console.log(`Subscriber 2: ${value}`)  
});

subject.next(1);
subject.next(2);

// Output:
// "Subscriber 1: 1"
// "Subscriber 2: 1"  
// "Subscriber 1: 2"
// "Subscriber 2: 2"

In this example, both subscribers receive the same values emitted by the subject.

A BehaviorSubject is a variant of Subject that requires an initial value and emits the current value to new subscribers. It‘s useful for representing a value that changes over time, like a settings configuration.

import { BehaviorSubject } from ‘rxjs‘;

const settingsSubject = new BehaviorSubject({ theme: ‘light‘ });

settingsSubject.subscribe(console.log);
// Output: { theme: ‘light‘ }

settingsSubject.next({ theme: ‘dark‘ });
// Output: { theme: ‘dark‘ }

settingsSubject.subscribe(console.log); 
// Output: { theme: ‘dark‘ }

When a new subscriber subscribes to the BehaviorSubject, it immediately receives the last emitted value. This makes it useful for cases where you want to cache and replay the latest value.

Testing Observables

Testing asynchronous code can be tricky, but observables make it easier by providing a declarative way to define expected behavior. RxJS provides a testing utility called marble diagrams that visually represent observable streams.

Marble syntax is a string that describes an observable stream over time. Here‘s the basic syntax:

-a--b-c--|

In this diagram, - represents a frame of time (typically 10 virtual milliseconds). a, b, and c represent emitted values. | indicates the observable completing.

Using marble diagrams, you can easily test observable scenarios:

import { TestScheduler } from ‘rxjs/testing‘;

const testScheduler = new TestScheduler((actual, expected) => {
  expect(actual).toEqual(expected);
});

testScheduler.run(({ cold, expectObservable }) => {
  const source$ = cold(‘-a--b-c--|‘, { a: 1, b: 2, c: 3 });
  const expected =     ‘-a--b-c--|‘;
  expectObservable(source$).toBe(expected);
});

In this test, we define the expected behavior of the observable using the marble diagram ‘-a--b-c--|‘. We then assert that the actual observable matches this expected behavior.

The TestScheduler allows us to manipulate time and control the speed of our tests. It‘s a powerful tool for testing complex observable scenarios.

Adoption and Performance

Reactive programming and observables have seen significant growth and adoption in recent years. According to the State of JavaScript 2020 survey, 27.8% of respondents have used RxJS, making it the 3rd most popular data layer library after Redux and Apollo Client.

Source: https://2020.stateofjs.com/en-US/technologies/datalayer/

One of the key benefits of observables is their performance. By leveraging operators like filter, distinctUntilChanged, and debounceTime, you can optimize data flows and minimize unnecessary computations and network requests.

For example, consider an autocomplete scenario where you want to fetch suggestions based on user input. With imperative programming, you might be tempted to make an HTTP request on every keystroke:

const searchInput = document.getElementById(‘search‘);

searchInput.addEventListener(‘input‘, () => {
  const term = searchInput.value;
  fetchSuggestions(term);
});

However, this can result in a flood of unnecessary network requests. With observables and RxJS operators, you can efficiently handle this:

import { fromEvent } from ‘rxjs‘;
import { map, filter, debounceTime, distinctUntilChanged, switchMap } from ‘rxjs/operators‘;

const searchInput = document.getElementById(‘search‘);
const inputObservable = fromEvent(searchInput, ‘input‘);

inputObservable.pipe(
  map(event => event.target.value),
  filter(term => term.length > 2),
  debounceTime(300),
  distinctUntilChanged(),
  switchMap(term => fetchSuggestions(term))
).subscribe(suggestions => {
  // Update UI with suggestions
});

In this optimized version, we use debounceTime to wait for a pause in typing before making a request, distinctUntilChanged to avoid duplicate requests, and switchMap to cancel previous requests if the user types something else. This results in a much more efficient and responsive user experience.

Observables Across Languages

While RxJS is the most popular implementation of observables for JavaScript, the concept of reactive extensions (ReactiveX) is not limited to a single language. ReactiveX is a cross-platform specification with implementations in various languages like Java, Python, C++, and more.

Having a consistent observable API across different languages allows for easier code sharing and architectural patterns. You can leverage the same concepts and operators whether you‘re working on the frontend with Angular or on the backend with Spring WebFlux.

// Java example with RxJava
Observable.just("Hello", "World")
    .subscribe(System.out::println);

# Python example with RxPY
from rx import of

source = of("Hello", "World")
source.subscribe(lambda value: print(value))

This cross-platform compatibility makes observables a powerful tool for building reactive systems that span multiple languages and runtimes.

Observables and Angular

Observables play a central role in Angular and are used extensively throughout the framework. Angular‘s HTTP client uses observables for all requests, allowing you to easily manipulate and compose response streams.

this.http.get(‘/api/data‘).pipe(
  map(data => data.items),
  catchError(error => of([]))
).subscribe(items => {
  this.items = items;
});

Angular‘s async pipe subscribes to an observable and automatically updates the view when new values are emitted. This declarative approach simplifies data binding and change detection.

<ul>
  <li *ngFor="let item of items$ | async">{{ item.name }}</li>
</ul>

Angular‘s reactive forms also leverage observables for form control values, status changes, and validation. This allows for real-time form updates and dynamic validation based on observable streams.

this.form = new FormGroup({
  name: new FormControl(‘‘),
  email: new FormControl(‘‘, [Validators.required, Validators.email])
});

this.form.valueChanges.pipe(
  debounceTime(500)
).subscribe(value => {
  console.log(value);
});

By embracing observables at its core, Angular provides a cohesive and reactive approach to building complex applications.

Conclusion

Observables are a powerful abstraction for working with asynchronous data streams. They offer a declarative and composable way to manage complex data flows, from user events to HTTP responses. With libraries like RxJS, you can leverage a rich set of operators to transform, filter, and combine observables, resulting in more concise and maintainable code.

As a full-stack developer, mastering observables is essential for building modern, reactive applications. Whether you‘re working on the frontend with Angular or on the backend with reactive frameworks, observables provide a consistent and efficient way to handle asynchronous operations.

By understanding the principles behind observables, leveraging the power of operators, and applying best practices like subscription management and testing, you can take your reactive programming skills to the next level.

So embrace the observable mindset, dive into the world of reactive programming, and start building more responsive, resilient, and scalable applications today!