Creating Your Own Operator
So far, you’ve used operators provided by the RxJS library. However, you shouldn’t be limited to those. With pipe, creating your own operators is easier than ever. An operator is a function that takes an observable, manipulates it in some way, and returns that new observable. This might be a bit complicated, so let’s look at some code. In this simple case, an operator is created that appends a string to each value passed through the observable.
| | import { Observable } from 'rxjs'; |
| | import { map } from 'rxjs/operators'; |
| | |
| ① | let stringAppendOperator = string => { |
| ② | return obs$ => { |
| | return obs$ |
| | .pipe(map(val => val + string)); |
| | }; |
| | }; |
| | |
| ③ | let myObservable$ = new Observable(o => { |
| | o.next('hello'); |
| | o.complete(); |
| | }); |
| | |
| | myObservable$.pipe( |
| | stringAppendOperator(' world!') |
| | ) |
| | .subscribe(next => { |
| ④ | console.log(next); |
| | }); |
- ①
-
Each of the operators you’ve seen so far is a function that returns another function. In this case, the operator is a function that takes a string, so that the end developer can define what should be appended to each value—in this case, world!
- ②
-
RxJS passes the observable to the inner function, and pipes it through the map operator to modify the value, appending the string.
- ③
-
To test this, an observable is created that emits a single value, then completes.
- ④
-
Finally, this should log ’hello world!’
This example cheats a bit by wrapping the map operator. Most operators included with RxJS create an entirely new observable, subscribe to the input observable, modifiy the value and return the newly-created observable. Here’s how the map operator works:
| | let mapOperator = someMappingFunction => { |
| | return obs$ => { |
| | return new Observable(o => { |
| | let subscription = obs$.subscribe( |
| | val => o.next(someMappingFunction(val)) |
| | ); |
| | () => subscription.unsubscribe(); |
| | }); |
| | }); |
| | }; |
Creating a new observable gets pretty complicated, and is probably overkill for any operator you want to write, but it’s good to have some understanding of what’s happening behind the scenes whenever you call an operator.
All this seems like a very messy way to do what we already learned how to do: manipulate a stream using the operators Rx provides. However, the power of creating your own operators comes from the ability to create reusable, composable chunks of functionality.
Remember the stock streaming example from Chapter 4, Advanced Async? Imagine we had multiple graphs on the page that all wanted to listen in on the stock filtering checkboxes. We could either copy and paste the same functionality for each graph, or create a single unit of functionality for that filtering and pass it into pipe:
| | function filterByCheckboxes(obs$) { |
| | return combineLatest( |
| | settings$, |
| | obs$, |
| | (enabledStocks, latestUpdates) => ({enabledStocks, latestUpdates}) |
| | ) |
| | .pipe( |
| | map(({enabledStocks, latestUpdates}) => { |
| | return latestUpdates |
| | .pipe( |
| | filter(stockHistory => enabledStocks.includes(stockHistory.name)) |
| | ); |
| | }) |
| | ); |
| | } |
| | |
| | stockPrices$ |
| | .pipe(filterByCheckboxes) |
| | .subscribe(updatePricesGraph); |
| | |
| | stockRatios$ |
| | .pipe(filterByCheckboxes) |
| | .subscribe(updateRatiosGraph); |
| | |
| | stockCaps$ |
| | .pipe(filterByCheckboxes) |
| | .subscribe(updateCapsGraph); |
So long as every stream ensures that the objects passed through have a name property that matches the stock ticker, this custom operator allows us to combine several operators together and reuse them across our applications. Now, what does this have to do with video games?