🏗️ How to create Custom Observable and Operators with RxJS and Node.js. 👀 With this example I want to show how you can generate custom Observables and Operators with RxJS and Node.js.

rxjs-custom-observable-and-operators.js

/**
 * Title: How to create Custom Observable and Operators with RxJS and Node.js.
 *
 * Description: With this example I want to show how you can generate custom
 * Observables and Operators with RxJS and Node.js.
 */

import { Observable } from 'rxjs';
import { request } from 'undici';

/** 
 * Create a custom Observable.
 * It contains a fetch call from which we receive a list of 100 todos.
 * { userId: number, id: number, title: string, completed: boolean }
 */
const observable$ = new Observable((observer) => {
  request('https://jsonplaceholder.typicode.com/todos')
    .then((res) => {
      res.body.json()
        .then(res => observer.next(res))
        .catch(err => observer.error(err));
    })
    .catch(err => observer.error(err));
});

/**
 * This method does exactly the same as map. I just leave it for you as an example.
 * @param {function} fn The function you want map to execute. 
 * @param {function} cb Callback function in order to return a result or error.
 */
const customOperatorMap = function (fn, cb) {
  return function (observable) {
    return new Observable((observer) => {
      observable.subscribe({
        next(value) {
          const resultMap = value.map(fn);
          observer.next(resultMap);
          cb(resultMap, null);
        },
        error(err) { observer.error(err); cb(null, err); },
        complete() { observer.complete() },
      });
    });
  }
}

/**
 * @description This method allows you to filter the received todos by id. 
 * @param {number} id Todo id to filter.
 * @param {function} cb Callback function in order to return a result or error.
 */
const customOperatorFilterById = function (id, cb) {
  return function (observable) {
    return new Observable((observer) => {
      observable.subscribe({
        next(value) {
          const resultFilter = value.filter(item => item.id === id);
          observer.next(resultFilter);
          cb(resultFilter, null);
        },
        error(err) { observer.error(err); cb(null, err); },
        complete() { observer.complete() }
      });
    });
  }
}

// Define a pipe with the two custom operators and the subscription to see the result.
observable$.pipe(
  customOperatorMap(item => ({ ...item, date: Date.now() }), (res, err) => {
    /**
     * This is the callback. You can do anything here.
     * ⚠️ In this cases, the correct is to work with tap to generate side effects,
     * this is only an example that you can creat anything.
     */
    // if (err) console.error(err);
    // console.log('customOperatorMap callback', res);
  }),
  customOperatorFilterById(99, (res, err) => {
    /**
     * This is the callback. You can do anything here.
     * ⚠️ In this cases, the correct is to work with tap to generate side effects,
     * this is only an example that you can creat anything.
     */
    // if (err) console.error(err);
    // console.log('customOperatorFilterById callback', res);
  }),
).subscribe(res => console.log(res));