Rx revisit

TLDR;

This post is a follow-up to my rx-java article. It briefly introduces the origins of Rx, its multi-threading implementation, and explains the important lift function used in its implementation.

Rx

Rx is essentially an advanced version of the Observer pattern. It wraps the subject being observed as an Observable (think of it as a collection that asynchronously produces elements), then registers callbacks for Observers via onNext, onError, and onCompleted for the appropriate scenarios. Notably, Rx provides operations such as filter, map/flatMap, merge, and reduce on Observables, making it as convenient to work with as a synchronous collection.

Origins

Note: This section references these slides

Netflix, as a major contributor to RxJava, ported Rx.Net to the JVM. Let’s trace the origins of RxJava through an evolution of Netflix’s architecture.

Netflix, as a well-known streaming media provider with 33 million subscribers, with peak downstream traffic accounting for 33% of North American internet traffic and 2 billion API requests per day, found that it was time to rethink their API architecture. Their previous architecture looked like this:

To reduce API call latency, engineers decided to move toward a coarse-grained API design:

A coarse-grained API design means business logic migrates to the server side, which inevitably results in parallel execution and nested calls on the server.

At the same time, we want API providers to hide their underlying concurrency implementation from callers. What does that mean?

Let’s look at a counterexample:

public Bookmark getBookmark(Long userId)

One day, a developer says: “This API takes too long — I don’t want to block the calling thread.” As the API provider, you’d naturally think of this solution:

public Future<Bookmark> getBookmark(Long userId)

Changing the return type to Future<Bookmark> prevents blocking the caller’s thread, while internally a new thread or thread pool handles the logic.

Then one day, the product manager says: “Requirements changed — we want to support multiple Bookmarks per user.” Happily, you update the API:

public Future<List<Bookmark>> getBookmark(Long userId)

At this point, the client-side developers who integrate with your API have had enough of the constant interface changes and are ready to come after you.

After nearly being done in, you finally ask:

Is there a data structure that can represent one or more results available in the future, delivering them to the caller only when ready, without the caller needing to know anything about the internal concurrency implementation?

Yes! Observable to the rescue

I mentioned the definition of Observable in my previous post. Let’s revisit it here:

Observable represents a consumable data collection (data provider). Consumers need not know whether data production is synchronous or asynchronous — it notifies consumers when data is available, when an error occurs, and when the data stream has ended.

In fact, Observable and Iterable are a pair of dual concepts — both represent collections of multiple elements, but they differ in key ways:

• Elements in an Observable can be produced asynchronously, whereas all elements in an Iterable must be available before they can be consumed.
• Observable is push-based: it notifies consumers when elements are available, when an error occurs, and when all elements have been consumed. Iterable is pull-based: consumers must actively poll for new elements, catch exceptions themselves, and handle completion themselves.

If we redesign the API above using Observable, it would look like this:

public Observable<Bookmark> getBookmark(Long userId)

Now regardless of whether synchronous return, asynchronous return, or asynchronous return of multiple Bookmarks is needed, the API signature never changes. Internally, the API can:

• Use the calling thread for computation, then deliver the result via onNext when done (native blocking call).

• Use a new thread or thread pool for computation, then deliver the result via onNext when done (satisfies async return requirement).

• Use multiple threads or a thread pool to fetch multiple bookmarks in parallel, calling onNext for each result (satisfies async multi-value return requirement).

Despite such significant internal changes, clients need not change anything. That’s the power of Observable’s elegant abstraction. This is the enormous benefit Rx brings, and the primary motivation for Netflix to port it to the JVM.

How It Works

Given how powerful Observable is, we naturally ask:

• Why can Observable be push-based?
• How does Observable support multiple concurrency implementations?

For the first question: Observable provides a subscribe method for clients to register callback functions. Observable then performs its computation and invokes the appropriate callbacks, making it appear as though the Observable is pushing results to its clients.

For the second question: Rx has an important concept called Scheduler — an abstraction over concurrency models. You can specify which concurrency model to use when creating an Observable. Let’s see how Scheduler abstracts over concurrency.

Scheduler

Rx’s default behavior is single-threaded. It is a free-threaded ^t1 model, meaning you can freely choose which thread executes a given task. If no scheduler is introduced when creating an Observable, the registered onNext, onError, and onCompleted callbacks will execute on the current thread (i.e., the thread where the Observable was created). The Scheduler provides a mechanism to specify on which thread callbacks will execute.

Scheduler Implementations:

• EventLoopsScheduler

  Maintains a pool of workers. When a new task is submitted via #schedule, a worker is selected via round-robin and the task is dispatched via Worker#scheduleActual.

• CachedThreadScheduler

  Uses a ConcurrentLinkedQueue to cache created threads for later reuse. Created threads have a time-to-live and are automatically removed when they expire.

• ExecutorScheduler

  Wraps an Executor instance and implements the Scheduler interface on top of it. Note that in this implementation, thread-hopping is unavoidable because the Scheduler has no knowledge of the thread behavior of the underlying Executor.

• ImmediateScheduler

  Executes tasks immediately on the current thread.

• TrampolineScheduler

  Assigns tasks to the current thread, but does not execute them immediately. Tasks are queued and executed after the current task completes.

Worker

The actual executor of callbacks. Internally backed by java.util.concurrent.ExecutorService for task execution, and also plays the role of a Subscription.

Summary

While studying the Principles of Reactive Programming on Coursera, I noticed that Erik Meijer jokingly said: “Don’t try to implement Observable yourself — just use the existing library.” Out of curiosity, I went ahead and read through the RxJava source code anyway, and came to appreciate just how elegant this model is. It brings a revolutionary experience to multi-threaded programming. If you’re interested in Rx.java, I strongly recommend reading through its source code.

Note: This should have been a separate blog post, but please forgive my laziness.

The lift Function

There is one function in Observable’s implementation that deserves special mention: lift.

Rx cleverly introduces a function type called Operator, which represents a mapping from one Subscriber to another Subscriber.

Many operations on Observable are implemented by defining an Operator and lifting it onto an Observable — for example, Observable#all, Observable#filter, Observable#finallyDo, and many others.

The signature of Observable#lift is as follows:

//inside Observable[T]
def lift[T, R](Operator[R, T]): Observable[R]

Introduction to the lift Function

Anyone with a functional programming background will likely be familiar with the name lift. As the name suggests, it takes a function that operates on a simple type and lifts it to work on a complex/container type.

Let’s look at a definition of lift:

Given two types A and B
and a function f: A => B
and a container M[_]
lift transforms f into a new function M[A] => M[B]

Thus, lift is defined as:

 def lift[A, B, M[_]](f: A => B): M[A] => M[B]

The only difference from the Observable#lift we saw above is that this lift returns a function. But on closer inspection, applying M[A] => M[B] to an M[A] instance produces the same effect as Observable#lift.

This sounds abstract, but once we substitute the symbols with familiar types, it becomes quite recognizable. Let’s apply the following substitutions step by step:

• A: String
• B: Int
• M: List[_]
• f: String => Int

Therefore, the type signature of lift becomes:

(String => Int) => (List[String] => List[Int])

Let’s use (s: String) => s.length as our f.

Given a list of strings xs: List[String],

lift(f)(xs) would give us the length of each string in xs.

Wait —

Isn’t that just xs.map(f)?

Yes! The map function is a common instance of lift.

Observable#lift

Now let’s see what role lift plays within Observable.

Before we begin, keep these type equalities in mind (:= means the two sides are equivalent types):

• Observable[_] := Subscriber[_] => Unit
• Operator[T, R] := Subscriber[R] => Subscriber[T]

Now let’s apply the type substitutions:

• A: Subscriber[R]
• B: Subscriber[T]
• M: Observable[_] (i.e. Subscriber[_] => Unit)
• f: Subscriber[R] => Subscriber[T]

Therefore, lift becomes:

(Subscriber[R] => Subscriber[T]) => (Subscriber[T] => Unit) => (Subscriber[R] => Unit)

Which is equivalent to:

(Subscriber[R] => Subscriber[T]) => (Observable[T] => Observable[R])

Given a ts: Observable[T] and a function f: Subscriber[R] => Subscriber[T], by applying lift, we obtain a result of type Observable[R].

t1: The opposite of the free-threaded model is the single-threaded apartment model, which means you must interact with the system through a designated thread.