RxJs Operator and Pipe for Angular

Operator in RxJS refers to several helper operation (map, filter, etc) used for processing input values emitted by an observable, and the output is a new observable with updated values. The input observable is not changed.

For example, the below method shows how an input observable gets processed and transformed into an output observable
function multiplyByTen(input) { var output = Rx.Observable.create(function subscribe(observer) { input.subscribe({ next: (v) => observer.next(10 * v), error: (err) => observer.error(err), complete: () => observer.complete() }); }); return output; } var input = Rx.Observable.from([1, 2, 3, 4]); var output = multiplyByTen(input); output.subscribe(x => console.log(x));


In the above code, the only thing may changed is how each data item will be handled in subscriber's next method, that is really how operator defines its function. In operator's map function, it defines how old data items are transformed into output observable's new data time as below :

    const nums = of(1, 2, 3);

    const squareValues = map(

        (val: number) => val * val

    );

    let obs: Observable<number> = squareValues(nums);

    obs.subscribe( v => {

      console.log(v);

    });

Similarly, the filter operator can be applied to an observable as below

    const odd = filter((n: number, ind: number) => {

      console.log('number ', ind, ' = ', n);

      return n % 2 === 0;

    });

    obs = odd(nums);

    obs.subscribe( v => {

      console.log(v);

    });

Essentially, the RxJs operator is a function that takes an input transform function as parameter, and returns a new function, which takes an observable as input parameter, and output a new observable as return value. Each value emitted from the input observable will be transformed or filtered by the transform handler, and then emitted to the output observable.

Multiple RxJs operators can also be piped together. Note, the RxJs pipe has nothing to do with angular string interpolation's pipe operation (|), and they just happen to have the same name.

The below is an example of using pipe

    const squareOddVals = pipe(

      filter((n: number) => n % 2 !== 0),

      map(n => n * n)

    );

    const squareOdd = squareOddVals(nums);

    squareOdd.subscribe( x => {

      console.log(x);

    });

Note, pipe is also a method of RxJS observable object, so it can be called directly on observable object as below

    const sqEven = of(1, 2, 3, 4, 5, 6)

      .pipe(

        filter((n: number) => n %2 === 0),

        map(n => n * n)

      ).subscribe( x =>

        console.log(x)

      );
Pipe has another important usage is flattening the embedded subscribe calls with exhaustMap, switchMap, concatMap and mergeMap.