Intro to RxJS Concepts with Vanilla JavaScript
Tags: javascript, rxjs, rx, beginners
Published: 2019-03-20 00:34:35

---
title: Intro to RxJS Concepts with Vanilla JavaScript
published: true
description: Breaking down observables, observers, and operators using vanilla JavaScript
tags: JavaScript, JS, RxJS, Rx, Beginner
---

A webinar recently inspired me that [egghead.io](egghead.io) hosted with [Andre Staltz](https://twitter.com/andrestaltz) and I wanted to share what I learned. Before the webinar, I was unfamiliar with RxJS and it was the first time I was exposed to the observer pattern. Before it was broken down, Observers seemed like magic. 

JavaScript has multiple APIs that use callback functions that all do nearly the same thing with slight variations.

**Streams**
```js
stream.on('data', data => {
   console.log(data)
})
stream.on('end', () => {
   console.log("Finished")
})
stream.on('error', err => {
   console.error(err)
})
```

**Promises**
```js
somePromise()
  .then(data => console.log(data))
  .catch(err => console.error(err))
```

**Event Listeners**
```js
document.addEventListener('click', event => {
  console.log(event.clientX)
})
```

The rough pattern you see is that there is an object, and inside the object, you have some method that takes a function, in other words, a callback. They're all solving the same problem, but in different ways, this causes you to have to carry the mental overhead of remembering the specific syntax for each of these APIs. That's where RxJS comes in. RxJS unifies all of this under one common abstraction.

So what even is an observable? It's an abstraction in the same way that arrays, functions, or objects are all abstractions. A promise can either resolve or reject, giving you back one value. An observable is capable of emitting values over time. You could consume streams of data from a server or listen for DOM events. 

**💀 Observable Skeleton**
```js 
const observable = {
  subscribe: observer => {

  },
  pipe: operator => {

  },
}
```

Observables are just objects that contain a `subscribe` and `pipe` method. Wait, what's going on here? What's an observer, or an operator? Observers are just objects that contain the callback methods for `next`, `error`, and `complete`. The `subscribe` method consumes an observer and passes values to it. So observable is acting as a producer, and the observer is its consumer. 

**👀 An Observer**
```js
const observer = {
  next: x => {
    console.log(x)
  },
  error: err => {
    console.log(err)
  },
  complete: () => {
    console.log("done")
  }
}
```

Inside of that `subscribe` method you pass some form of data to the observer's methods. 

**Subscribe Method**
```js 
const observable = {
  subscribe: observer => {
    document.addEventListener("click", event => {
      observer.next(event.clientX)
    })
  },
  pipe: operator => {

  },
}
```

Here we are just listening for clicks made anywhere in the document. If we ran this and made a call to `observable.subscribe(observer)`, we would see the x coordinates of your clicks showing up in the console. So what about this `pipe` method? The `pipe` method consumes an operator and returns a function, and makes a call to the resulting function with the observable.  

**Pipe Method**
```js 
const observable = {
  subscribe: observer => {
    document.addEventListener("click", event => {
      observer.next(event.clientX)
    })
  },
  pipe: operator => {
    return operator(this)
  },
}
```

Cool but what's an operator? Operators are for transforming your data. Arrays have operators, like `map`. `map` lets you take a step back and run some function over everything in the array. You could have an array and then another array that is a mapped version of the first.

Let's write a `map` function for our observable. 

**🗺️ Map Operator**
```js 
const map = f => {
  return observable => {
    subscribe: observer => {
      observable.subscribe({
        next: x => {
          observer.next(f(x))
        },
        error: err => {
          console.error(err)
        },
        complete: () => {
          console.log("finished")
        }
      })
    },
    pipe: operator => {
      return operator(this)
    },
  }
}
```

A lot is going on here so let's break it down.

```js
const map = f => {
  return observable => {
```

Here we are passing in a function and returning a function that expects an observable. Remember our `pipe` method? 

```js
pipe: operator => {
  return operator(this)
},
```

To run the operator on the observable, it needs to get passed into `pipe`. `pipe` is going to pass the observable it's called on into the function that our operator returns. 

```js
subscribe: observer => {
  observable.subscribe({
```

Next, we are defining the `subscribe` method for the observable that we are returning. It expects an observer, which it receives in the future when `.subscribe`  gets called on the returned observable, either through another operator or explicitly. Then, a call gets made to `observable.subscribe` with an observer object.

```js
{
  next: x => {
    observer.next(f(x))
  },
  error: err => {
    console.error(err)
  },
  complete: () => {
    console.log("finished")
  }
}
```

In the observer's `next` method you can see that a call to a future observer's `next` is made with the function that we originally passed into `map` and an `x` value passed into `next`. Let's run our new `map` operator on our observable!

```js
observable
  .pipe(map(e => e.clientX))
  .pipe(map(x => x - 1000))
  .subscribe(observer)
```

That final `subscribe` is needed or none of the operations inside of those operators execute, that's because they are all wrapped up in their observer's `subscribe` methods. In those `subscribe` methods is a call to `subscribe` the previous observer in the chain, but the chain has to begin somewhere. 

So let's follow what happens when this runs. 

1. The first pipe gets called on observable, `map` gets curried with `this`
2. `map` is called with `e => e.clientX` and it returns a function
3. The function gets called with the original `observable` and an observable gets returned
   1. **We'll call it observable2**
4. `pipe` is called on `observable2` and curries `map` with `this`
5. `map` is called with `x => x - 1000` and it returns a function
6. That function gets called with `observable2` and an observable gets returned
   1. **We'll call it observable3**
7. `.subscribe` gets called on `observable3` with an observer passed in
8. `.subscribe` gets called on `observable2` with the operator's observer passed in
9. `.subscribe` is called on the original observable with the operator's observer passed in
10. A click event happens with a `clientX` of `100`
11. `observer2.next(100)` gets called 
12. `observer3.next(100)` gets called
13. `observer.next(-900)` gets called and logs `-900` to the console. 
14. Done!

You can see the chain happen here. When you call `subscribe` you are asking for information, each link asks the previous link in the chain for it until it reaches the data and the `next` method from its observer gets called. That data then rises back up the chain, getting transformed along the way, until it then reaches the final observer. 
  
Here is the code in its entirety. 

```js
const observable = {
  subscribe: observer => {
    document.addEventListener("click", event => {
      observer.next(event.clientX)
    })
  },
  pipe: operator => {
    return operator(this)
  }
}

const observer = {
  next: x => {
    console.log(x)
  },
  error: err => {
    console.log(err)
  },
  complete: () => {
    console.log("done")
  }
}

const map = f => {
  return observable => {
    subscribe: observer => {
      observable.subscribe({
        next: x => {
          observer.next(f(x))
        },
        error: err => {
          console.error(err)
        },
        complete: () => {
          console.log("finished")
        }
      })
    },
    pipe: operator => {
      return operator(this)
    },
  }
}
```