Reading the RxJS 6 Sources: Map and Pipe

This series consists of my personal notes as I work through the RxJS source code. A summary of the key points will be provided at the end, so don’t feel overwhelmed by the details.

Welcome back. Today I’m eager to explore the inner workings of pipe in RxJS. This article starts with a high-level look at how map and pipe function, followed by a deep dive into the source code.

Recap

In the previous installment, I examined the of method for creating an observable. I’ll build on that simple Stackblitz example, this time enabling map and pipe. Prior knowledge from the earlier article isn’t required to follow along. Here’s the snippet from Stackblitz:

Let’s dive into map!

You can access the Stackblitz here.

Before examining the sources, it’s helpful to understand map and pipe conceptually. Having a high-level picture prevents getting lost in the complexities of the implementation.

Here’s what I know going in:

• map is an operator that transforms data by applying a function
• pipe composes operators (such as map, filter, etc.)

Understanding Map

The primary role of map is transformation

map is a straightforward operator. It accepts a projection function and applies it to every value originating from the source observable.

In this scenario, the observable returned by of('World') serves as the source, and the single value 'World' is passed through pipe to map's projection function, which is defined as:

x => `Hello ${x}!` // projection function
// It's used like this:
of('World').pipe(map(x => `Hello ${x}!`));

The projection function takes 'World' as its input parameter x and produces the string Hello World!. map encapsulates the projection function within an observable, which then emits the resulting string Hello World!. Keep in mind, operators invariably return observables.

I’ve covered the basics of map and other operators in detail in another article. Some of that content will be revisited here. Essentially, if you grasp how Array.prototype.map operates, most of that knowledge transfers to observables.

We’ll revisit map in more depth shortly. Now, let’s turn our attention to pipe.

Exploring Pipe

pipe is the main focus of this article. Unlike map, which is an operator, pipe is a method on Observable used for composing operators. pipe was introduced in RxJS v5.5 to transform code that looked like this:

of(1,2,3).map(x => x + 1).filter(x => x > 2);

into this:

of(1,2,3).pipe(
  map(x => x + 1),
  filter(x => x > 2)
);

The output and concept—composing operators—are identical, but the syntax differs. pipe provides several advantages:

• It tidies up Observable.prototype by removing operators
• It enhances the tree-shakeability of the RxJS library
• It simplifies the creation and use of third-party operators (eliminating the need to patch Observable.prototype).

Brief Detour (skip if you’re familiar with pipe)

If pipe for composition is new to you, it’s worth seeing how it works with regular functions before applying it to operators. Let’s examine a simplified version of pipe that operates on ordinary functions:

const pipe = (...fns) => 
           initialVal => 
           fns.reduce((g,f) => f(g), initialVal);

In this example, pipe is a function that takes functions as arguments. These arguments are gathered into an array called fns using ES6 rest parameters (…fns). pipe then returns a function that accepts an initialValue, which is passed into reduce. This value is fed into the first function in fns, and its output is then passed to the second function, and so on—creating a pipeline. For instance:

const pipe = (...fns) => initialVal => fns.reduce((g,f) => f(g), initialVal);
const add1 = x => x + 1;
const mul2 = x => x * 2;

const res = pipe(add1,mul2)(0); // mul2(add1(0)) === 2

You can experiment with a basic pipe at this Stackblitz link.

In RxJS, the concept is to construct a pipeline of operators (like map and filter) to apply to each value emitted by a source observable, such as of(1,2,3).

This strategy enables the creation of small, reusable operators like map and filter, which can be combined as needed using pipe.

Composition is a fascinating subject that I can’t fully cover here. If you’re interested, I recommend Eric Elliott’s series on the topic.

Let’s Dig into the Sources!

I’ll start by inserting a debugger statement inside map. This provides access to map within the dev tools debugger and a way to step up to pipe.

In the dev tools, I see:

Now that I’m positioned in the call stack, I can explore further.

Notice that Observable.subscribe initiates everything in the call stack. Since observables are typically lazy, data won’t flow through pipe and map until we subscribe to the observable.

var sub = source.subscribe(...)

Inside map, I spot MapOperator and MapSubscriber, which appear noteworthy:

Don’t worry about the specifics

On line 55, source refers to the observable generated by of('World'). It gets subscribed to on line 56, which triggers it to emit its single value, 'World', and then complete.

On line 56, a MapSubscriber instance is created and passed to source.subscribe. We’ll see later that the projection function gets invoked within MapSubscriber’s _next method.

On line 56, this.project is the projection function passed to map:

For now, this.thisArg can be disregarded. Thus, line 56 accomplishes the following:

return source.subscribe(new MapSubscriber(subscriber, this.project, this.thisArg));

1. It calls subscribe on source, which is the observable returned by of('World').
2. The observer (next, error, complete, etc.) passed into source.subscribe will be the Subscriber returned by MapSubscriber, which takes the current subscriber and the projection function as arguments.

As a quick note, this pattern is prevalent among RxJS operators. In fact, they all seem to follow this template:

• Export a public function, such as map, filter, or expand.
• Export a class implementing Operator, like MapOperator. This class implements the Operator interface’s call method, subscribing to the source observable, e.g., return source.subscribe(new MapSubscriber(…));. This ties the observables into a subscriber/observer pipeline.
• A class extending Subscriber. This class implements methods such as _next. It houses the unique logic for each operator. For instance, in map, the projection function is called within MapSubscriber’s _next. In filter, the predicate is invoked inside FilterSubscriber’s _next, and so forth.

I’ll demonstrate how to craft your own operator in a future article (though it’s often simpler to compose existing ones with pipe). Meanwhile, the RxJS sources offer a helpful guide here.

Anyway, back to the debugging session.

Eventually, once subscribe is called, MapSubscriber._next gets triggered.

Observe that the projection function, project, supplied to map, is called on line 81. The result (in this case 'Hello World!') is then returned and passed to this.destination.next(result) on line 86.

Stepping into this.project.call puts us in the lambda passed to the map call

This clarifies how map applies the projection function to each value from the source observable upon subscription. That covers this step. If another operator existed in the chain, the observable from map would feed into it.

This illustrates data flow through a single operator. But how does it flow through multiple operators…

Revisiting Pipe

To understand that, I need to examine pipe. It’s invoked on the observable returned by of('World').

pipeline

On line 331, pipeFromArray is called with operations, an array of all operators passed to pipe. In this case, it’s just the lone map operator:

operations could contain many, many operators

The function returned from pipeFromArray(operations) is invoked with this, pointing to the observable from of('World').

With only one operator (map), line 29 returns it directly.

Line 33 is noteworthy. It’s where all operators passed to pipe are combined using Array.prototype.reduce. It doesn’t trigger when only one operator is supplied (likely for performance reasons).

Let’s examine a more intricate example with multiple map operators.

Working with Multiple Maps

With a grasp of what map and pipe do, I’ll try a more involved example. This time, I’ll use map three times!

Hello World of RxJS

The key difference is that pipe will utilize reduce this time:

The input variable remains the observable from of('World').

By stepping through each function in fns as reduce calls it, I can observe the string being built incrementally as it passes through each map operator, ultimately producing Hello World of RxJS.

Each subscriber’s _next function is invoked in sequence

Understanding data flow through a single operator makes it easy to extend that concept to multiple operators.

Map meets filter

Let's bring filter into the picture, just to illustrate that map isn't special in any way. The expectation is that every operator follows the same structural blueprint.

The console will display 3 and 4

Here, of(1,2,3) creates an observable that, once subscribed to, pushes three distinct values — 1, 2, and 3 — before calling complete. These values travel through the pipeline sequentially. map increments each by one and then forwards the new values one after another on the observable it produces. filter then subscribes to the output of map, evaluating each incoming value against its predicate ( x => x > 2 ). It emits an observable that only lets through values exceeding 2. In this scenario, that means 3 and 4.

For a deeper look into how the subscriber chain links operators together through subscription, I've got an in-depth write-up here.

Wrapping up

• Operators such as map and filter are functions that both accept and return observables.
• Each operator proffers a public-facing function — map, filter, and the like — which is what gets imported from 'rxjs/operators' and handed over to pipe.
• Beneath every operator sits a *Operator class implementing the Operator interface, enabling it to subscribe to upstream observables.
• Each operator also has a *Subscriber class housing its core logic — be it the projection function for map, the predicate function for filter, or something else.
• We've observed how pipe chains operators together, internally folding the source values across the operator list.

Up next, I'll tackle more advanced map variants and explore how higher-order observables are put together.

Still curious?

For those keen on constructing a basic observable from the ground up, I've put together a compact series:

Part 1: Arrays
Part 2: Containers
Part 3: Creating Observables with of, from, and fromEvent
Part 4: Operators