Angular: RxJS 구독을 선언적으로 관리

소개
토론을 위해 Angular 애플리케이션에서 RxJS를 보는 것이 더 일반적이기 때문에 예제에 Angular를 사용하겠습니다.

구독 관리는 애플리케이션 성능에 매우 중요합니다. Observable을 구독하면 Observable에 콜백 함수를 등록하고 Observable은 콜백 목록을 유지합니다. 논리가 완료되었을 때 구독을 취소하지 않으면 메모리 누수가 발생할 수 있습니다.

예를 들어 보겠습니다. ngOnInit 에서 다른 종류의 관찰 가능 항목을 구독하는 것이 일반적입니다.

ngOnInit () {
  this.service.users$.subscribe(nextCb, errorCb, completeCb)
 }

하지만 다른 경로로 이동하여 이 구성 요소로 돌아간다면 어떻게 될까요? 당신은 몇 번이고 구독할 것입니다.

누군가 "음 구독을 변수에 저장하고 ngOnDestroy에서 구독을 취소하겠습니다"라고 말할 것입니다.

users$

ngOnInit () {
  this.users$ = this.service.users$.subscribe(nextCb, errorCb, 
  completeCb)
 }
ngOnDestry(){
  this.users$.unsubscribe()
}

기술적으로는 맞지만 구독이 여러 개인 경우에는 어떻게 됩니까? 상황이 정말 빨리 지저분해질 것입니다.

ngOnDestry(){
  this.variable1$.unsubscribe()
  this.variable2$.unsubscribe()
  this.variable3$.unsubscribe()
....
}

RxJS oberatortakeUntil는 구독을 선언적으로 제거하는 데 유용할 수 있습니다.
| 통지자 Observable이 값을 내보낼 때까지 원본 Observable이 내보낸 값을 내보냅니다.

@Component({ ... })
export class AppComponent implements OnInit, OnDestroy {
  destroy$: Subject<boolean> = new Subject<boolean>();

  constructor(private service: Service) {}

  ngOnInit() {
    this.service.users$
    .pipe(takeUntil(this.destroy$))
    .subscribe(({data}) => {
      console.log(data);
    });

 this.productsService.products$
    .pipe(takeUntil(this.destroy$))
    .subscribe(({data}) => {
      console.log(data);
    });
  }

  ngOnDestroy() {
    this.destroy$.next(true);
    this.destroy$.unsubscribe();
  }
}

이것이 최선의 방법입니까? 실제로 함수 내에서 관찰 가능 항목을 구독해야 하는 경우 매우 좋은 트릭입니다. 최적의 경우 Angular 파이프를 사용하여 async에서 구독을 처리하도록 해야 합니다. 이것은 더 이상 .ts 파일을 구독할 필요가 없기 때문에 매우 유용합니다.

다음은 Deborah Kurata's github 의 예입니다.

export class ProductListComponent {
  pageTitle = 'Product List';
  private errorMessageSubject = new Subject<string>();
  errorMessage$ = this.errorMessageSubject.asObservable();

  private categorySelectedSubject = new BehaviorSubject<number>(0);
  categorySelectedAction$ = this.categorySelectedSubject.asObservable();

  products$ = combineLatest([
    this.productService.productsWithAdd$,
    this.categorySelectedAction$
  ])
    .pipe(
      map(([products, selectedCategoryId]) =>
        products.filter(product =>
          selectedCategoryId ? product.categoryId === selectedCategoryId : true
        )),
      catchError(err => {
        this.errorMessageSubject.next(err);
        return EMPTY;
      })
    );

  categories$ = this.productCategoryService.productCategories$
    .pipe(
      catchError(err => {
        this.errorMessageSubject.next(err);
        return EMPTY;
      })
    );

  vm$ = combineLatest([
    this.products$,
    this.categories$
  ])
    .pipe(
      map(([products, categories]) =>
        ({ products, categories }))
    );

  constructor(private productService: ProductService,
              private productCategoryService: ProductCategoryService) { }

  onAdd(): void {
    this.productService.addProduct();
  }

  onSelected(categoryId: string): void {
    this.categorySelectedSubject.next(+categoryId);
  }
}

이전 예에서 사용자는 범주를 선택하고 이 범주의 제품을 볼 수 있습니다. .subscribe() 파일에 단일.ts이 없는 이 모든 논리. 모든 구독은 마운트 해제될 때 자동으로 구독을 취소하는 템플릿의 파이프async로 처리됩니다.