How to create custom pipe operators in RxJS

RxJS is a popular library for reactive programming in JavaScript, and creating custom pipe operators can help simplify complex data transformations. In this article, we'll explore how to create a custom pipe operator in RxJS.

February 18, 2023 rxjs

Creating custom pipe operators by mixing existing operators

RxJS provides a powerful set of built-in operators that can be used to transform, filter, and combine streams of data.

Filter null and undefined values – filter in practice

In RxJS, filter is a higher-order function that only emits items from the original sequence that satisfy a specified predicate function.

1
import { filter, of, OperatorFunction } from "rxjs";
2


3
function isDefined<T>(): OperatorFunction<T, Exclude<T, null | undefined>> {
4
  return filter((value): value is Exclude<T, null | undefined> => value !== null && value !== undefined);
5
}

Thanks of T & {}, null and undefined types will be removed from types.

1
of(false, null, true, undefined).subscribe((value) => console.log(value));
2
//                                            ^? value: boolean | null | undefined
3


4
of(false, null, true, undefined)
5
  .pipe(isDefined)
6
  .subscribe((value) => console.log(value));
7
//              ^? value: boolean

P.S: T & {} means exactly the same like Exclude<T, null | undefined>.

Filter passed values as an argument – advanced usage of filter operator

Instead of hardcoded null and undefined values in the function you can move it to the function’s argument.

1
import { filter, OperatorFunction } from "rxjs";
2


3
function isNot<T, G extends unknown[]>(...forbiddenValues: G): OperatorFunction<T, Exclude<T, G[number]>> {
4
  return filter((value): value is Exclude<T, G[number]> => {
5
    for (const forbiddenValue of forbiddenValues) {
6
      if (value === forbiddenValue) {
7
        return false;
8
      }
9
    }
10


11
    return true;
12
  });
13
}

Thanks to Exclude<T, G[number]> TypeScript can exclude defined types.

1
of(false, null, true, undefined).subscribe((value) => console.log(value));
2
//                                           ^? value: boolean | null | undefined
3


4
of(false, null, true, undefined)
5
  .pipe(isNot(null, false as const, undefined))
6
  .subscribe((value) => console.log(value));
7
//            ^? value: true

Operator for auto-complete – combine three operators into one

Auto-complete, or word completion, is a feature in which an application predicts the rest of a word a user is typing. It is very important to limit the number of requests to a service to prevent excessive usage and protects the service from being overloaded or overwhelmed.

See the example of how to create a custom auto-complete operator in RxJS to protect your service from redundant calls.

1
import { MonoTypeOperatorFunction, pipe, fromEvent } from "rxjs";
2
import { auditTime, distinctUntilChanged, filter, map } from "rxjs/operators";
3


4
function limitEvents<T extends string>(): MonoTypeOperatorFunction<T> {
5
  return pipe(
6
    auditTime(500),
7
    distinctUntilChanged(),
8
    filter((value) => value.length > 3),
9
  );
10
}
11


12
fromEvent(document.querySelector("input")!, "change")
13
  .pipe(
14
    map((event) => event.target.value),
15
    limitEvents(),
16
  )
17
  .subscribe();

auditTime ignores source values for a given time in milliseconds then emits the most recent value from the source. Thanks to that each change will emit value only once per 500 milliseconds.

distinctUntilChanged emit value to the next operator only when a value has been changed.

filter will emit value only if the input has more than 3 characters.

Logger operator – single operator with predefined values

The purpose of a logger is to help developers understand the behavior of a system, diagnose issues, and troubleshoot problems. See the example of how to create a custom logger operator in RxJS to log each event to the console.

1
import { MonoTypeOperatorFunction } from "rxjs";
2
import { tap } from "rxjs/operators";
3
import { interval } from "rxjs";
4


5
function logger<T>(tag: string = "DEFAULT"): MonoTypeOperatorFunction<T> {
6
  return tap({
7
    next: (value) => console.log(`%c[${tag}] %cNEXT`, "color: #bada55", "color: #FFF", value),
8
    error: (error) => console.error(`%c[${tag}] %cERROR`, "color: #bada55", "color: #F00", error),
9
    complete: () => console.log(`%c[${tag}] %cCOMPLETED`, "color: #bada55", "color: #0F0"),
10
    subscribe: () => console.log(`%c[${tag}] %cSUBSCRIBE`, "color: #bada55", "color: #FF0"),
11
    unsubscribe: () => console.log(`%c[${tag}] %cUNSUBSCRIBE`, "color: #bada55", "color: #FF0"),
12
  });
13
}
14


15
interval(500).pipe(logger("INTERVAL")).subscribe();

Thanks to the defined colors in the log console, the output will look colorful and intuitive.

Creating a custom pipe operator based on a custom implementation

Creating a custom pipe operator should be the last step to resolving your problem. Try to resolve a problem using available operators, if it is not possible only then try to implement a custom operator.

To create a custom pipe operator, we first need to define a function that takes an observable as input and returns a new observable. This function should contain the logic for the transformation we want to perform.

1
import { Observable } from "rxjs";
2


3
function isDefined<T>(source: Observable<T>): Observable<T & {}> {
4
  return new Observable((subscriber) => {
5
    const subscription = source.subscribe({
6
      next(value) {
7
        if (value !== undefined && value !== null) {
8
          subscriber.next(value);
9
        }
10
      },
11
    });
12


13
    return subscription;
14
  });
15
}

You should always tear down created subscriptions inside a new Observable. To do it return created subscription or add a custom callback which will be called when the source will be unsubscribed.

1
import { Observable } from "rxjs";
2


3
function interval(period: number): Observable<number> {
4
  return new Observable((subscriber) => {
5
    let i = 0;
6
    const interval = setInterval(() => subscriber.next(i++), period);
7


8
    return () => clearInterval(interval);
9
  });
10
}

Thanks of T & {}, null and undefined types will be removed from types.

1
of(false, null, true, undefined).subscribe((value) => console.log(value));
2
//                                           ^? value: boolean | null | undefined
3


4
of(false, null, true, undefined)
5
  .pipe(isDefined)
6
  .subscribe((value) => console.log(value));
7
//             ^? value: boolean

P.S: T & {} means exactly the same like Exclude<T, null | undefined>.

Custom pipe operator with arguments

To create a custom pipe operator with custom arguments, we need to create a function generator.

1
import { Observable, of, OperatorFunction } from "rxjs";
2


3
function isNot<T, G extends unknown[]>(...forbiddenValues: G): OperatorFunction<T, Exclude<T, G[number]>> {
4
  return function (source) {
5
    return new Observable((subscriber) => {
6
      const subscription = source.subscribe({
7
        next(value) {
8
          for (const forbiddenValue of forbiddenValues) {
9
            if (value === forbiddenValue) {
10
              return;
11
            }
12
          }
13


14
          subscriber.next(value as Exclude<T, G[number]>);
15
        },
16
      });
17


18
      return subscription;
19
    });
20
  };
21
}

Do not forget to tear down your subscription!

Thanks to Exclude<T, G[number]> TypeScript can exclude defined types.

1
of(false, null, true, undefined).subscribe((value) => console.log(value));
2
//                                           ^? value: boolean | null | undefined
3


4
of(false, null, true, undefined)
5
  .pipe(isNot(null, false as const, undefined))
6
  .subscribe((value) => console.log(value));
7
//             ^? value: true

Conclusion

Creating a custom pipe operator in RxJS can help simplify complex data transformations and make your reactive programming code more readable and reusable, but remember, creating your own pipe operator from scratch should be your last choice.

Do you like the content?

Your support helps me continue my work. Please consider making a donation.

Buy Me A Coffee

Donations are accepted through PayPal or Stripe. You do not need a account to donate. All major credit cards are accepted.

Leave a comment