Angular 리액티브 프로그래밍과 RxJS

Observable (데이터 스트림 생산 : 생산자)
            ↑     |
    구독     |     |       방출
 subscribe  |     |   notification
            |     ↓
┌─────────────────────────────────┐
│                                 │
│ Observer1, Observer2, Observer3 │
│                                 │
└─────────────────────────────────┘

// myComponent.component.ts
import { Observable } from '@angular/core';
...
export class MyComponent implements OnInit {
  ngOnInit() {

    // 옵저버블이 '구독'될 때 호출되는 구독 함수
    const subscriber = (observer) => {
      // next()           - next()
      // error()          - error()
      // complete()       - complete()
      // return () => {}  - 구독 해지될 때
    }

    // 옵저버블 생성
    const observable$ = new Observable(subscriber);

    // 구독
    observable$.subscribe(
      next(), // next 에 반응할 메서드
      error(), // error 에 반응할 메서드
      complete(), // complete 에 반응할 메서드
    );
  }
}

// Observable.ts
/**
   * @constructor
   * @param {Function} subscribe the function that is called when the Observable is
   * initially subscribed to. This function is given a Subscriber, to which new values
   * can be `next`ed, or an `error` method can be called to raise an error, or
   * `complete` can be called to notify of a successful completion.
   */
constructor(subscribe?: (this: Observable<T>, subscriber: Subscriber<T>) => TeardownLogic) {
    if (subscribe) {
      this._subscribe = subscribe;
    }
  }

// Observable.ts
subscribe(next?: (value: T) => void, error?: (error: any) => void, complete?: () => void): Subscription;

// ...
const subject = new Subject();
const hotObservable = subject.asObservable(); // hotObservable 생성
subject.next(1); // 데이터 방출
hotObservable.subscribe(...) // 구독

// mycomponent.component.ts
...
ngOnInit() {
  const numbers$ = new Observable((observer) => {
      observer.next(1);
      observer.next(2);
      observer.next(3);
    });

    numbers$.subscribe(
      (value) => console.log(`1st next: ${ value }`),
      (error) => console.error(`1st error: ${ error }`),
      () => console.log('1st complete')
    );

    numbers$.subscribe(
      (value) => console.log(`2nd next: ${ value }`),
      (error) => console.error(`2nd error: ${ error }`),
      () => console.log('2nd complete')
    );
}
/*
1st next: 1
1st next: 2
1st next: 3
2nd next: 1
2nd next: 2
2nd next: 3
*/

// mycomponent.component.ts
...
ngOnInit() {
  const subject = new Subject();
    const numbersBySubject$ = subject.asObservable();

    subject.next(1);
    subject.next(2);
    subject.next(3);

    numbersBySubject$.subscribe(
      (value) => console.log(`1st next: ${ value }`),
      (error) => console.error(`1st error: ${ error }`),
      () => console.log('1st complete')
    );

    numbersBySubject$.subscribe(
      (value) => console.log(`2nd next: ${ value }`),
      (error) => console.error(`2nd error: ${ error }`),
      () => console.log('2nd complete')
    );

    subject.next(4);
    subject.next(5);
    subject.next(6);
}
/*
1st next: 4
2nd next: 4
1st next: 5
2nd next: 5
1st next: 6
2nd next: 6
*/

// mycomponent.component.ts
...
ngOnInit() {
  ngOnInit(): void {
    // COLD
    console.log('COLD');
    const coldObservable$ = Observable.create((observer) => observer.next(Math.random()));
    coldObservable$.subscribe(
      (value) => console.log(`1st Cold Observable's observer: ${value}`),
    );
    coldObservable$.subscribe(
      (value) => console.log(`2nd Cold Observable's observer: ${value}`),
    );

    // HOT
    console.log('HOT');
    const subject = new Subject();
    const hotObservable$ = subject.asObservable();
    hotObservable$.subscribe(
      (value) => console.log(`1st Hot Observable's observer: ${value}`),
    );
    hotObservable$.subscribe(
      (value) => console.log(`2nd Hot Observable's observer: ${value}`),
    );
    subject.next(Math.random());
  }
}

/*
COLD
1st Cold Observable's observer: 0.3458584261362281
2nd Cold Observable's observer: 0.6683291493239305

HOT
1st Hot Observable's observer: 0.10435603985906838
2nd Hot Observable's observer: 0.10435603985906838
*/

// ...
export class MyComponent implements OnInit, OnDestroy {
  myArray = [1, 2, 3, 4, 5];
  subscription: Subscription;

  ngOnInit() {
    const observable$ = from(this.myArray);
    this.subscription = observable$
      .pipe(
        // 옵저버블 변형
        map(item => item * 2), // 2, 4, 6, 8, 10
        filter(item => item > 5), // 6, 8, 10
        tap(item => console.log(item)) // print 6, 8, 10
      )
      .subscribe(
        ...
      )
  }

  ngOnDestroy() {
    this.subscription.unsubscribe(); // 구독 해제 (메모리 누수 방지)
  }
}