[Angular] Observable

✅ 개요

• 옵저버블은 이벤트 처리, 비동기 프로그래밍, 여러 값을 연달아 처리할 때 사용
• 옵저버 패턴은 옵저버(observer) 라는 의존성 객체들의 목록을 관리하고, 이 옵저버가 구독(subject) 하고 있다가 상태가 변화하는 것을 감지하는 소프트웨어 디자인 패턴
• Angular 앱은 RxJS 라이브러리를 활용함
  
  

✅ 개념 정의

• 값을 발행하는 함수는 소스(Source)
• Source 함수는 누군가 이 옵저버블을 구독하는 쪽에서 옵저버블의 subscribe() 메서드를 실행해야 실행됨
• 옵저버블이 전달하는 값을 받는 객체를 구독자(subscriber) 라고 하며,
• 구독자는 옵저버블이 종료되거나 에러가 발생할 때, 구독을 종료할 때까지 옵저버블이 보내는 값을 계속 받음
• 옵저버블은 기본 자료형, 메시지, 이벤트 등 다양한 종류의 값을 여러번 보낼 수 있음
• 그래서 연속된 키 입력, HTTP 응답, 타이머가 주기적으로 실행되는 것은 옵저버블을 활용하기 좋은 예임
• 옵저버블로 전달되는 값은 RxJS 연산자로 처리
• 옵저버블로 전달되는 값을 단계적으로 변환하거나, 여러 옵저버블 소스를 하나로 조합하는 기능은 RxJS가 제공하는 강력한 기능
  
  

✅ Observer 알림과 예시

next	옵저버블로 전달되는 값을 처리합니다. 옵저버블 구독이 시작된 후 한번도 실행되지 않을 수 있고, 여러번 실행될 수도 있습니다.
error	옵저버블에서 에러가 발생한 것을 처리합니다. 옵저버블에서 에러가 발생하면 옵저버블 인스턴스가 종료되며 구독도 종료됩니다.
complete	옵저버블이 종료된 것을 처리합니다. 이 핸들러가 실행되고 나면 next나 error는 더이상 실행되지 않습니다. 옵저버블 구독은 자동으로 종료됩니다.

// 이 함수는 subscribe()가 실행될 때 같이 실행됩니다.
function sequenceSubscriber(observer: Observer<number>) {
  // 1, 2, 3을 순서대로 보내고, 옵저버블을 종료합니다.
  observer.next(1);
  observer.next(2);
  observer.next(3);
  observer.complete();

  // 이 함수는 모든 데이터를 동기 방식으로 보내기 때문에
  // 구독이 해제될 때 실행해야 하는 로직은 없습니다.
  return {unsubscribe() {}};
}

// 구독자 함수를 인자로 전달하면서 새로운 Observable 인스턴스를 만듭니다.
const sequence = new Observable(sequenceSubscriber);

// 옵저버블을 실행하고 스트림이 전달될 때마다 로그를 출력합니다.
sequence.subscribe({
  next(num) { console.log(num); },
  complete() { console.log('Finished sequence'); }
});

// 로그:
// 1
// 2
// 3
// Finished sequence

✅ RxJS 라이브러리

• Promise를 옵저버블로 변환하기

import { from, Observable } from 'rxjs';

// Create an Observable out of a promise
const data = from(fetch('/api/endpoint'));
// Subscribe to begin listening for async result
data.subscribe({
  next(response) { console.log(response); },
  error(err) { console.error('Error: ' + err); },
  complete() { console.log('Completed'); }
});

• 카운터를 옵저버블로 변환하기

import { interval } from 'rxjs';

// Create an Observable that will publish a value on an interval
const secondsCounter = interval(1000);
// Subscribe to begin publishing values
const subscription = secondsCounter.subscribe(n =>
  console.log(`It's been ${n + 1} seconds since subscribing!`));

• 이벤트를 옵저버블로 변환하기

import { fromEvent } from 'rxjs';

const el = document.getElementById('my-element')!;

// Create an Observable that will publish mouse movements
const mouseMoves = fromEvent<MouseEvent>(el, 'mousemove');

// Subscribe to start listening for mouse-move events
const subscription = mouseMoves.subscribe(evt => {
  // Log coords of mouse movements
  console.log(`Coords: ${evt.clientX} X ${evt.clientY}`);

  // When the mouse is over the upper-left of the screen,
  // unsubscribe to stop listening for mouse movements
  if (evt.clientX < 40 && evt.clientY < 40) {
    subscription.unsubscribe();
  }
});

• AJAX 요청을 옵저버블로 변환하기

import { Observable } from 'rxjs';
import { ajax } from 'rxjs/ajax';

// 옵저버블을 생성하고 AJAX 요청을 보냅니다.
const apiData = ajax('/api/data');
// Subscribe to create the request
apiData.subscribe(res => console.log(res.status, res.response));

✅ Async Pipe와 사용하기

• AsyncPipe는 옵저버블이나 Promise를 구독하고, 이 객체가 담고 있는 마지막 값을 반환
• 그리고 새로운 값이 전달되면 컴포넌트가 변화를 감지하도록 알림
• 아래 예제는 컴포넌트의 뷰에서 옵저버블 타입인 time 프로퍼티를 바인딩하는 예제
• 이 옵저버블은 컴포넌트에서 새로운 스트림을 생성할 때마다 계속 갱신됨

@Component({
  standalone: true,
  selector: 'async-observable-pipe',
  template: `<div><code>observable|async</code>:
       Time: {{ time | async }}</div>`,
  imports: [CommonModule]
})
export class AsyncObservablePipeComponent {
  time = new Observable<string>(observer => {
    setInterval(() => observer.next(new Date().toString()), 1000);
  });
}