DEV Community: Joseph Avery Ferrante

RxJS - Simplifying Complex Operator Chains

Joseph Avery Ferrante — Tue, 10 Mar 2020 21:01:10 +0000

RxJS is a powerful library that allows multiple operators to be applied to data coming from the Observable stream. While the library greatly simplifies working with asynchronous data, it can still introduce complex and hard to read code. In this article we will explore a strategy for simplifying a chain of complex operations into a format that is easier to understand.

Assumptions

Familiarity with RxJS, Observables, and using the pipe() function to apply multiple operators to data emitted through the stream.

Base Problem

Imagine we have an Observable stream where we want to transform the data that is coming through:

interval(500).pipe(
    filter((num: number) => num % 2 === 0),
    take(10)
    map((num: number) => num * 10),
);

Here we use the interval function to emit values every 500 milliseconds.
Looking at the operators in the pipe function, you can derive that we are

Allowing only even numbers to pass
Taking a total of 10 values before completing
Multiplying each number by 10

While this example may be simple and contrived, you must realize that understanding what is happing within the pipe() function takes effort and time. If you extract this simple example to a real world use case with multiple operators performing application logic, it can quickly become complex and take quite a bit of effort, especially for new members, to grasp what is happening with the data.

Solution - Version 1

We can actually create small, bite sized, and well named functions to house our transformation logic. Let's focus on the first operator in the pipe chain:

filter((num: number) => num % 2 === 0)

We can refactor this into an encapsulating function:

private function takeEvenNumbers() {
    return filter((num: number) => num % 2 === 0);
}

which can then be used within the original observable chain:

interval(500).pipe(
    this.takeEvenNumbers(),
    take(10)
    map((num: number) => num * 10),
);

Already this simplifies the process of introducing someone to the logic within the operator chain, but we can go further.

Solution - Version 2

Just as we can have multiple operators in a pipe() function attached to the Observable, we can also return multiple operators from our encapsulating functions. Let's refactor the first two operators from our original pipe chain:

private function take10EvenNumbers() {
    return pipe(
        filter((num: number) => num % 2 === 0),
        take(10)
    );
}

NOTE: the pipe function used here is imported from 'rxjs' (import { pipe } from 'rxjs')

We can now rewrite the original Observable as such:

interval(500).pipe(
    this.take10EvenNumbers(),
    map((num: number) => num * 10),
);

Depending on the data transformation happening/level of granularity the developer desires, you can use these self created operator functions to build up other operator functions:

private function takeEvenNumbers() {
    return filter((num: number) => num % 2 === 0);
}

private function take10EvenNumbers() {
    return pipe(
        this.takeEvenNumbers(),
        take(10)
    );
}

interval(500).pipe(
    this.take10EvenNumbers(),
    map((num: number) => num * 10),
);

Solution - Version 3

Looking at the previous solution, while improved, is potentially too rigid or specific. The function take10EvenNumbers(), while useful here, could be generalized for use elsewhere. We can achieve such:

private function takeXEvenNumbers(amount: number) {
    return pipe(
        filter((num: number) => num % 2 === 0),
        take(amount)
    );
}

We now have flexibility, allowing us to take any amount of even numbers.

interval(500).pipe(
    this.takeXEvenNumbers(10),
    map((num: number) => num * 10),
);

Conclusion

Using the method described above, we can abstract potentailly complex & confusing logic into bite sized and understandable chunks. The onus for deciding what granularity/abstraction level is useful falls to the developer.

interval(500).pipe(
    this.takeXEvenNumbersAndMultiplyBy(10, 10)
);

The above may or may not be useful, but that is an exercise for the development team. As a final, more real world example, imagine facing this:

this.clientService.getServersByDealerId(dealerId).pipe(
      pluck('results'),
      mergeMap((objArr: Server[]) => timer(2000).pipe(mapTo(objArr))),
      mergeMap((objArr: Server[]) => {
        let observableArray = [];
        objArr.forEach(client => {
          observableArray.push(this.livestreamService.getMediaSourcesByServerId(client.id).pipe(
            map(objArr => {
              objArr.forEach(o => o.hostClientId = client.name)
              return objArr
            })
          ))
        })
        return forkJoin(observableArray);
      }),
      map((camerasInServers: Camera[][]) => {
        return camerasInServers
          .filter(sub => sub.length !== 0)
          .map(sub => {
            let container: CameraContainer = {
              hostClientId: sub[0].hostClientId,
              cameras: sub
            }
            return container;
          });
      }),
      distinctUntilChanged((p: CameraContainer[], q: CameraContainer[]) => JSON.stringify(p) === JSON.stringify(q))
    )

versus facing this

this.clientService.getServersByDealerId(dealerId).pipe(
    pluck('results'),
    this.emitServerResultsEvery(2000),
    this.getCamerasFromServers(),
    this.mapToCameraContainer(),
    this.emitChangedCameras()
)

The second is easier to read, debug, and understand what is going on. Reading the second one you may derive relatively quickly what is happening: a request for server objects that contain IDs needed for polling child object changes.

RxJS & the combineLatest Operator "Gotcha"

Joseph Avery Ferrante — Mon, 09 Mar 2020 22:25:08 +0000

Today, I spent a good chunk of my workday debugging an issue I had with an observable stream spawned from the combineLatest function. I want to share this life lesson in hopes others do not suffer a similar fate.

I've boiled the problem down to the code snippet below:

// Will output one value, an array of 3 numbers
const oneValue$ = of([10, 20, 30]);
// Will output two values
const twoValues$ = of(100, 200);

combineLatest(oneValue$, twoValues$).pipe(
  map(([val1, val2]) => {
    return {
      a: val1.splice(0, 1)[0],
      b: val2
    }
  })
).subscribe(output => console.log(output));
// Desired Output:
// { a: 10, b: 100 }
// { a: 10, b: 200 }

// Actual Output:
// { a: 10, b: 100 }
// { a: 20, b: 200 }

Notice the desired vs. actual outputs. Let's walk through this example:

We have two observable streams, one that will emit only 1 value in its lifetime, the other will emit two values.

Using combineLatest I want to construct a new return value based on the latest values from each stream.

I map the two emitted values [val1, val2] to a new object.

I want this new object to have two properties where

property a is the first value in the array from oneValue$ (i.e. the value 10)
property b is the latest value in twoValues$ (i.e. the value 100, then 200)

So I want my output to look like

  // Desired
  { a: 10, b: 100 }
  { a: 10, b: 200 }

  // Actual
  { a: 10, b: 100 }
  { a: 20, b: 200 }

My a value isn't always the first value from the array. Strange?

combineLatest should re-emit the last value from oneValue$ when twoValues$ emits the second time. That should always be the array [10, 20, 30]. I'm using the function splice(0, 1)[0] to grab that first item in the array, yet the second emission seems to have grabbed the second item of the array.

Have you figured it out yet?

The splice() function is permanently altering the array values

combineLatest is emitting the same array value during the second emission, but
The splice() function actually alters the values of the array during each mapping. Remember, JavaScript is always passing by reference. In our map() function, val1 is another variable that is pointing to the same array object being held in the combineLatest operator. We then modify the array contents. So when combineLatest passes the array reference to map() the second time, we receive the same (modified) array reference.

Obvious fixes are to use the spread operator if it's an array or object of primitive values, or possibly pulling in something like cloneDeep from the lodash library for more complex objects.

Many array functions return new arrays without altering the original, however, not all do. Make sure you make a conscious decision around this!