Angular subscribe elegant

I'd like to ask about my subscribe method. I want to get my objects(employees) from firebase and do a loot of operations on my objects, like avg salaries, avg missed days... In my opinion it looks so horrible. My service void:

 getEmployees(): Observable<Employee[]> {
    return this.db.list<Employee>(this.Api_url).snapshotChanges()
      .pipe(map(response => response.map(car => this.assignKey(car))));
  }

And my ngOnInit:

  ngOnInit() {

    this.subscribtion = this.mainservice.getEmployees().subscribe((emps)=> {
      this.employees = emps;
      this.workersdatas = [];
      this.educationdatas = [];
      this.checkavaragesalaries(emps);
      this.countMonthAbsences(emps, 1);
      this.countMonthSalaries(emps, 1);
      this.departments.forEach((one)=> {
        const dep = emps.filter(on => on.department.includes(one) && this.FilterEmployees(on)).length;
        this.workersdatas.push({name: one, value: dep});
      });

      this.educations.forEach((one)=> {
        const edu = emps.filter(on => on.education.includes(one)).length;
        this.educationdatas.push({name: one, value: edu});
      });

      const mynumb =this.educationdatas.map(on => on.value);
      this.mosteducation = this.educationdatas.filter(one => one.value === Math.max(...mynumb))[0].name;
    }); }

Should I unsubscribe() it on ngOnDestroy or that's not necessary? Is that allowed to write it like that?

My favourite unsubscribe pattern is to use takeUntil. takeUntil() accepts an observable, and will stop the subscription when that observable emits a value.

The long-winded way to do this is emit a value in ngOnDestroy().

destroyed = new Subject<void>();

ngOnInit(): void {
  myService.getData().pipe(
    takeUntil(this.destroyed)
  ).subscribe(data => {
    // process data
  });
}

ngOnDestroy(): void {
  this.destroyed.next();
  this.destroyed.complete();
}

This gets quite tedious once you've done this a few times. The best solution I've seen for this (and currently use) is to set up a shared function that emits the destroyed value for you. Source Credit

import { OnDestroy } from '@angular/core';

import { ReplaySubject, Subject, Observable } from 'rxjs';

export function componentDestroyed(component: OnDestroy): Observable<void> {
  const oldNgOnDestroy: () => void = component.ngOnDestroy;
  const destroyed: Subject<void> = new ReplaySubject<void>(1);
  component.ngOnDestroy = () => {
    oldNgOnDestroy.apply(component);
    destroyed.next(undefined);
    destroyed.complete();
  };
  return destroyed.asObservable();
}

It is then almost trivial to use in your component:

ngOnInit(): void {
  myService.getData().pipe(
    takeUntil(componentDestroyed(this))
  ).subscribe(data => {
    // process data
  });
}

ngOnDestroy(): void {}

All you have to do is implement ngOnDestroy in your component and add takeUntil(componentDestroyed(this)) to your pipe.