How to Read the RxJS 6 Sources Part 1: Understanding of() and Subscriptions

Why Reading the RxJS Sources is Worth It

RxJS holds a special place in my heart. The way observables bring a declarative, composable style to both asynchronous and synchronous coding is something I genuinely enjoy. Once everything clicked for me, there was no going back. Working with RxJS is simply a pleasure.

Still, the inner workings of the library have long felt like a black box. I’ve put together a short series on building a custom observable from the ground up. If you’re looking for the foundations, that’s a good place to start.

That said, those articles didn’t dive into the RxJS codebase itself.
Having spent time reading through the Angular and React internals, it feels like the right moment to tackle my favorite library as well.

Who This is For

Do you enjoy RxJS? Are you curious about its implementation, interested in how things work under the hood, or simply have some spare time to kill?

My personal goal is to start contributing to RxJS soon, so getting familiar with the source is a necessary step.

How This Series Works

Think of these articles as my personal notes while I work through the sources. I’ll aim to keep things concise, so the content stays navigable.

At the end of the day, these are just my observations. I’m sharing them in case anyone wants to tag along on this exploration. Each post will finish with a recap, so you can skip the nitty-gritty if that’s not your style.

If you’re short on time, jump to the summary at the end.
(although the journey is where the fun lives)

I should be upfront: I’m not an RxJS authority. Much of what I write here is educated guesswork. For definitive answers, the core team is your best resource.

Getting Started

Whenever I explore a new codebase, I prefer to start with a simple example, set some breakpoints, and trace through the execution. This naturally leads me into various functions and classes, which helps me piece together the library’s architecture.

Let’s begin with the standard RxJS starter on Stackblitz.
https://stackblitz.com/edit/read-rxjs-sources

map and pipe are coming soon

For now, I’m going to strip things down by commenting out pipe and map. The only thing happening is that the string 'World' gets wrapped in an observable using the of operator. What we get is a cold observable that emits 'World' upon subscription, and then completes.

I’ve placed a debugger statement on line 12 inside the observer’s next callback to give us a good inspection point. Here’s the stack trace that appears when line 12 is hit:

ye gods man

Don’t squint at that just yet. It basically screams, “there’s a lot to uncover” ?

Let’s work from the top. Since observables are lazy, nothing runs until subscribe is invoked on line 11. So Observable.subscribe is naturally the first function in our call stack.

Observable.subscribe in Detail

Here’s the relevant source code straight from RxJS:

 subscribe(observerOrNext?: PartialObserver<T> | ((value: T) => void),
            error?: (error: any) => void,
            complete?: () => void): Subscription {

  const { operator } = this;
  const sink = toSubscriber(observerOrNext, error, complete);

  if (operator) {
    operator.call(sink, this.source);
  } else {
    sink.add(
      this.source || (config.useDeprecatedSynchronousErrorHandling && !sink.syncErrorThrowable) ?
      this._subscribe(sink) :
      this._trySubscribe(sink)
    );
  }

  if (config.useDeprecatedSynchronousErrorHandling) {
    if (sink.syncErrorThrowable) {
      sink.syncErrorThrowable = false;
      if (sink.syncErrorThrown) {
        throw sink.syncErrorValue;
      }
    }
  }

  return sink;
}

Take a look at the arguments for subscribe. It accepts an observer (comprised of next, error, complete), which can be provided in two ways:

• as three separate functions: next, error, and complete
• as an object holding those three functions as properties

This is likely what the PartialObserver type defines, so I’ll make a mental note to explore that later.

Also worth noting: subscribe gives back something typed as Subscription.

For the moment, lines 5–16 are the focus. Lines 18–24 appear to handle an edge case.

Line 5: destructuring pulls the operator property from this. Here’s what this looks like in my scenario:

There’s no operator property, so { operator } resolves to undefined. I’ll loop back to that. It’s a good bet that operator gets set when we start chaining operators.

On line 6 I step into toSubscriber, handing it the observer (whether that’s three functions or an object. In this instance, I passed just one function).

In my case, I gave subscribe a single function, so the nextOrObserver parameter holds that function (the next callback from the stackblitz example). I didn’t supply anything for error or complete. That leads me to the last line of toSubscriber.

return new Subscriber(nextOrObserver, error, complete)

I believe the purpose of toSubscribe is to normalize the observer input and return a Subscriber instance. Let’s step inside:

toSubscriber

The call to super invokes the constructor for Subscription, which looks like this:

I’ll map out these class relationships later. For now, keep in mind that Subscriptions are a big deal in RxJS. They are the disposable resources that enable data consumption.

Back in subscriber.ts, we end up returning a SafeSubscriber instance. SafeSubscriber is something I’ve run into quite a few times in the RxJS codebase, so I’ll set aside a dedicated article for it later. It also extends Subscription. At this point, it feels like all this logic decides which subscription flavor to return based on the observer type passed into Observable.subscribe…it’s starting to get tangled. Time to steer back to the main storyline.

Returning to subscribe:

The sink variable holds the subscriber, which in this case is just a Subscriber, with a destination of type SafeSubscriber.

subscriber.add performs this action

duckies!! (163)

I’ll dive deeper into add below. The main point is that Observable.subscribe hands back a Subscriber, which gives us the ability to unsubscribe at will. I’m fairly confident operators can tap into this feature too (think takeUntil).

That covers the basics here. Next, let’s look at what surrounds the subscribe call.

Subscriptions and Their Class Structure

First, let’s look at Subscription.

According to the source:

/**
Represents a disposable resource, such as the execution of an Observable. A Subscription has one important method, `unsubscribe`, that takes no argument and just disposes the resource held by the subscription. Additionally, subscriptions may be grouped together through the `add()` method, which will attach a child Subscription to the current Subscription.
When a Subscription is unsubscribed, all its children (and its grandchildren)will be unsubscribed as well.
*/

That’s intriguing. From my experience building observables manually, chains like of(1,2,3).map(x => x + 1).filter(x > 2) mean each operator subscribes to the one before it. So filter subscribes to the observable from map, which subscribes to the one from of, and so forth. This implies that unsubscribing from the top-level observable (filter) effectively tears down everything. Maybe that’s the job of the add method?

The unsubscribe method marks closed as true and clears some other references to null. Those specifics can wait for now; no need to get lost in the weeds.

As for add, the source explains:

Adds a tear down to be called during the unsubscribe() of this Subscription. Can also be used to add a child subscription.
If the tear down being added is a subscription that is already unsubscribed, is the same reference `add` is being called on, or is`Subscription.EMPTY`, it will not be added.
If this subscription is already in an `closed` state, the passed
tear down logic will be executed immediately.
When a parent subscription is unsubscribed, any child subscriptions that were added to it are also unsubscribed.

‘teardown’ points to something of type TearDownLogic. Likely a function meant to run when we unsubscribe. Time to verify that!

I swapped of for interval(1000) so we get a value every 1000ms, and I trigger unsubscribe after 5000ms. As expected, the lambda passed to sub.add fires on unsubscribe. It also runs when the source observable completes (like with of).

So add is clear now: it lets you register logic that runs on unsubscription or on completion.

*Note: RxJS Core Team member Nicholas Jamieson wrote an outstanding piece on Subscribers and add here.

At this point, I feel comfortable with Subscription, so let’s move forward.

---

The Internals of of()

Now, I’ll set a breakpoint inside of to see what happens before subscription.

The of() creational method

With our example, ...args is simply ['World'], given that’s the only value we provided.

Lines 76–81 are about schedulers, which dictate an observable’s timing (sync or async, and if async, whether microtask, macrotask, or animation frame). Schedulers are on my list to study eventually, but they’re not needed right now. It’s standard practice to pass a scheduler as the last argument to creational functions like of, which is what’s happening on line 76.

Since args.length equals 1 here, I can see line 86 will run. Let’s examine scalar(args[0] as T).

When in doubt, always return a new Observable

Compact functions like this are refreshing when reading through code ?

With just one value (‘World’), it’s treated as a scalar. On line 4, a new Observable is created (but not executed — that laziness again). This Observable takes a subscriber, like any other. subscriber refers to the observer object we examined earlier (next, error, complete).

Lines 5 and 6 show that the subscriber’s next method gets called with ‘World’, and then it completes (since a scalar is a single value).

Lines 8 and 9 handle some cleanup, and on line 10, the observable is returned. So the observable defined on line 4 is what you get when you write:

let source = of('World'); //of returns the observable from line 4

This notion is absolutely fundamental to grasping observables. I can’t emphasize that enough.

Observable creators and operators, without exception, return observables.

Why does that matter? Without it, composing observables would be impossible. You couldn’t chain things like of(1,2,3).map(x=>x+1).filter(x=>x>2). That chaining of of, map, and filter is composition, made possible because each returns an Observable (which exposes methods like map and filter). Previously, these operators lived on Observable.prototype, but the team has shifted away from that to improve tree-shaking. That’s where pipe comes in. I’ll get into the modern pipe syntax later in this series.

After of hands us that observable, we subscribe to it, and we’re back to familiar territory. Observable.subscribe triggers, and the magic on line 205 kicks in:

on 205, this._trySubscribe(sink) will eventually get to the `next` callback

That call to next lands on the subscriber passed into the observable created by scalar!

value is ‘World’. A few more objects get traversed, but ultimately, subscriber.next is the callback I initially passed to subscribe:

We’re home!

And there you have it. We’ve traced how of creates an observable and how the data eventually flows into the observer’s next method. Along the way, we saw the string 'World' get pulled out of the observable ?

By the way, my favorite variable name ever:

love the sense of humor ?

?

Multiple arguments and of()

Next, I want to see how fromArray behaves, so I’ll pass two strings, 'Hello' and 'World', into of:

At line 88, args holds ['Hello', 'World'] while scheduler remains undefined. Time to step into fromArray.

Since !scheduler evaluates to true, a fresh Observable gets returned—same pattern as scalar in the previous example. The difference: this time subscribeToArray(['Hello', 'World']) will be called. What does that function actually do? (Observe that subscribeToArray gets invoked right here with [‘Hello’, ‘World’] as its argument—keep that in mind.)

Lines 10–23 can be skipped for now, though they look promising. That add function shows up again, and this time it appears to handle scheduling tasks—irrelevant here because I didn’t pass a scheduler.

Here’s the implementation of subscribeToArray:

What we have is a function that accepts an ArrayLike—here, [‘Hello’, ‘World’]—and gives back a function taking a subscriber (an observer, essentially). That inner function loops through each element, invokes subscriber.next(array[i]), and then calls complete on line 12 once iteration finishes. It’s basically scalar but calling next once per array value.

Don’t forget: subscribeToArray merely returns that subscriber-accepting function, which then gets passed into the observable produced on line 8 of fromArray. Nothing flows through the observable yet. Only when I subscribe to the observable coming out of fromArray does it know to walk through [‘Hello’, ‘World’] and trigger next on whatever observer I supply, for every item. So here, next fires twice, followed by complete.

That takes care of scalars and arrays with of. One scenario remains.

Nested of() calls

Time to nest observables.

Nothing surprising happens in this case. The inner of(‘World’) produces a scalar observable, which immediately becomes the argument for the outer of call, as shown here:

Once again, scalar gets invoked, wrapping the observable that contains 'World'.

Eventually, the value reaching our next handler is the nested observable itself:

That’s because of has no flattening logic for observables. Later articles will explore operators like flatMap that handle this.

Key takeaways

Here’s what stood out from this walkthrough:

• of consistently returns an observable
• the number of values passed to of determines whether scalar or fromArray builds that observable
• the observable returned takes a subscriber—the argument given to subscribe—which can be three functions (next, error, complete) or an object with those as properties; only next is strictly necessary
• of won’t unpack nested observables
• subscriptions deserve their own deep dive later
• schedulers do too

Tracing a seemingly trivial example through the source code takes more effort than you’d expect. In the next installment, I’ll activate map and pipe to see what unfolds. Stay tuned—and let me know if these deep dives are helpful or enjoyable.

Additional reading

For building a simple observable from the ground up, check out my mini-series:

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