How I reverse engineered RxJs and Learned Reactive Programming?

Yes, the title is not a typo. We are actually going to reverse engineer RxJs (tons of code to come ;) ). But before we proceed let me tell you why I embarked on this crazy endeavor.

As programmers, we are curious by nature. I work with reactive libraries such as RxJs and React.js everyday. However, one bright morning I got curious about how these frameworks leveraging reactive programming under the hood.

Do I really understand what reactive programing is? How does it actually work? I asked myself.

After a weekend of digging through blog posts and reading books I sort of got the concept. However, I figured reverse engineering something would be a great way to actually nail down the concepts, so I decided to reverse engineer RxJs.

A Quick Intro:

Reactive programming is programming with asynchronous data streams.

The Data streams could be anything, it could be user inputs (i.e. position of your mouse, click events), it could be streams of data from a server (i.e. your twitter feed, or real time data from a socket server). Our Application will react according to these streams of data.

For example as you are receiving twitter feeds in realtime in your application state will change. Maybe you want to put the most popular tweets on top. So your application is subscribed to the incoming streams of data and it reacts to the data and puts the most popular tweet on top. In brief this concept of subscribing to data streams and changing the application accordingly is reactive programing.

Are you bored? Trust me this is not going to be one of those blog posts with lots of concepts. We will be diving into code now.

Let’s build a class called Observablesince it is the most fundamental building block of RxJs.

class Observable {
  constructor() {
    this.fnArray = [];
  }

  subscribe() {}

  emit() {}
}

const o = new Observable();

Alright we just created a basic class called Observable with two methods. We initialized an empty list called fnArray. This array will hold all our subscribed objects.
Let’s implement the subscribe method first. This method will take in a function as an argument and push it in our fnArray.

subscribe(fn) {
    this.fnArray.push(fn);
}

Now let’s implement the emit function as well. The job of the emit function is to loop over the fnArray and execute those functions one after another.

emit(v) {
  for (let fun of this.fnArray) {
    fun(v);
  }
}

We can also replace that for loop with a map. But why tho? Well that's what the cool kids in JS land are doing apparentnow.and curry functions are kindda cool!! so let’s do that now.

emit(v) {
-  for (let fun of this.fnArray) {
-    fun(v);
-  }
+  this.fnArray.map(fun => fun(v))
}

Okay, now let’s use our newly created class.

function printFunction(thing) {
 console.log(`I will print the ${thing}`)
}

const o = new Observable();
o.subscribe(printFunction);

First we created a function printFunction that prints whatever variable passed in. We initialized a new observable instance and called the subscribe method on it and passed in our printFunction as argument.

Remember the printFunction will be stored in the fnArray. Now what do you think will happen if we call the emit method? Let’s try

o.emit("Apple");
o.emit("Orange");
o.emit("Pear");

This gives us the following output

I will print the Apple
I will print the Orange
I will print the Pear

Okay now we are able to subscribe to a function or event and emit something based on that function. Here’s how the entire code looks like so far.

class Observable {
  constructor() {
    this.fnArray = [];
  }

  subscribe(fn) {
    this.fnArray.push(fn);
  }

  emit(v) {
    this.fnArray.map(fun => fun(v));
  }
}

function printFunction(thing) {
  console.log(`I will print the ${thing}`);
}

const o = new Observable();
o.subscribe(printFunction);

o.emit("Apple");
o.emit("Orange");
o.emit("Pear");

Now let’s get into the interesting bits. We can subscribe to multiple functions. For example we can do something like this

o.subscribe(x => console.log(x * 2));
o.subscribe(x => console.log(x + 2));

o.emit(4)

which returns

// 8
// 6

because our emit call looped over all the function in that array of function that is initialized on the class constructor.

notice that I am using arrow function now. We are also able to compose our functions with any combinations we wish.

const square = num => num * num;
o.subscribe(x => printFunction(x * 2));
o.subscribe(x => printFunction(square(x)));
o.emit(4);

// outputs

// I will print the 8
// I will print the 16

In the first scenario we composed our function with printFunction. In the second scenario we created a square function and composed it with printFunction.

This is sort of cool isn’t it?
Alright we can compose functions but we need a better way to compose them. Something more comprehensive like pipe in RxJS. So let’s build that mechanism.

const pipe = (f, g) => x => g(f(x));

We defined a new function called pipe that takes 2 functions as arguments and returns a function that takes a parameter then returns the composed function of f of g.
What did just happen
We took 2 functions as argument. Then we took another value as argument and we apply first function f with value x. Then we took the return value of f(x) and applied function g.
This could be a little confusing, if you are I highly recommend you do some reading on currying function in JavaScript.
Using the pipe function now we can do something like this

o.subscribe(
 pipe(
   square,
   printFunction,
 )
)
o.emit(4);

// outputs
// I will print the 16

But we have a problem here. We want to be able to pass in any numbers of functions and then we should be able to compose them. So if we have f,g,h,k ⇒ k(h(g(f))).

So we'll modify our pipe like this

const pipe = (...funcs) => x => funcs.reduce((effects, f) => f(effects), x);

what functional magic is this? Well, first of all we are taking in a number of functions with our spread operator. (...funcs) part specifies that we can take in any number of functions in order. Then we are taking in a value x as an argument to operate on. funcs.reduce will go over each functions and return the updated value of x and pass it in the next function that’s in the series. Think of this as a series execution. At the end of our execution x is still the same because we do not mutate values in pure functions.

More on pure function in my next article so stay tuned 😊

Now let me show you why we did this. Let’s take a look at the code below

o.subscribe(
 pipe(
   square,
   double,
   square,
   printFunction
 )
);
o.emit(2);

// outputs
// I will print the 64

You see now can compose functions without really caring much for their orders, and we can also keep the data immutable.
However, our implementation is missing one thing. We can not collect our data in between pipe. What I mean by this is, we can not break and collect our value after the second double is applied. RxJs has a tap method that allows this. So let’s go and implement a tap method.

const tap = fun => x => {
 fun(x);
 return x;
};

For this method we take in a function and a value and we apply function with the value and
return the original value. This way now we can tap and take out values in a specific position of the pipe stream.

o.subscribe(
 pipe(
   square,
   double,
   tap(printFunction),
   square,
   printFunction
 )
);
o.emit(2);

// outputs
// I will print the 8
// I will print the 64

This is pretty much it. We technically have the barebone functionality of a reactive library like RxJS. *** Now I want to show you a practical implementation of our reactive library***.

So let’s say we have some incoming asynchronous data. (i.e. mouse pointer position could be an example) and based on that data I want to perform some state change in my application. So this is how we will handle this with our reactive library

o.subscribe(pipe(
 filter(x => {
   if(x > 0) {
     console.log('In Range')
     return x;
   }
   console.log('Out of Range')
   return 0
 }),
 square,
 tap(printFunction),
));

o.emit(2);
o.emit(-4);
o.emit(8);
o.emit(4);
// outputs
// In Range
// I will print the 4
// Out of Range
// I will print the 0
// In Range
// I will print the 64
// In Range
// I will print the 16

So we can do this funnel like data filtering with our library just like RxJS. I hope this gave you some insight on how RxJS operates behind the scene.

In Part 2 we will further break down all the operations of RxJS. Stay tuned, leave a comment if you have any feedback, follow me for more articles like this

⚡️⚡️⚡️⚡️⚡️⚡️

Enjoying the ride so far? head over to part 2 🕟 🕔 🕠 🕕.

How I Reverse Engineered RxJs and Learned Reactive Programming? Part 2

Shadid Haque ・ Dec 3 '19 ・ 4 min read

#javascript #todayilearned #functional #webdev