The RxJS Library

리 액티브 프로그래밍은 데이터 스트림 및 변경 전파와 관련된 비동기 프로그래밍 패러다임 (Wikipedia)입니다. RxJS (JavaScript 용 Reactive Extensions)는 비동기식 또는 콜백 기반 코드 (RxJS Docs)를보다 쉽게 작성할 수 있도록 관찰 성을 사용하는 리 액티브 프로그래밍 용 라이브러리입니다.

RxJS는 Observable 유형의 구현을 제공합니다.이 유형은 유형이 언어의 일부가되고 브라우저가 지원할 때까지 필요합니다. 라이브러리는 관측 가능 객체를 생성하고 작업하기위한 유틸리티 함수도 제공합니다. 이러한 유틸리티 함수는 다음과 같은 경우에 사용할 수 있습니다.

• 비동기 작업에 대한 기존 코드를 observables로 변환
• 스트림의 값 반복
• 값을 다른 유형으로 매핑
• 스트림 필터링
• 여러 스트림 작성

RxJS는 새로운 관측 가능 객체를 생성하는 데 사용할 수있는 여러 가지 기능을 제공합니다. 이러한 함수는 이벤트, 타이머, 약속 등과 같은 것들로부터 관찰 대상을 만드는 프로세스를 단순화 할 수 있습니다. 예 :

import { from } 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
secondsCounter.subscribe(n =>
  console.log(`It's been ${n} seconds since subscribing!`));

import { fromEvent } from 'rxjs';
 
const el = document.getElementById('my-element');
 
// Create an Observable that will publish mouse movements
const mouseMoves = fromEvent(el, 'mousemove');
 
// Subscribe to start listening for mouse-move events
const subscription = mouseMoves.subscribe((evt: MouseEvent) => {
  // 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();
  }
});

import { ajax } from 'rxjs/ajax';
 
// Create an Observable that will create an AJAX request
const apiData = ajax('/api/data');
// Subscribe to create the request
apiData.subscribe(res => console.log(res.status, res.response));

연산자는 observables 기반을 기반으로 컬렉션의 정교한 조작을 가능하게하는 함수입니다. 예를 들어, RxJS는 map (), filter (), concat () 및 flatMap ()과 같은 연산자를 정의합니다.

연산자는 구성 옵션을 사용하고 소스를 관찰 할 수있는 함수를 반환합니다. 이 리턴 된 함수를 실행할 때, 연산자는 소스 observable의 방출 값을 관찰하고,이를 변환하고, 변환 된 값의 새로운 관측 가능 값을 리턴합니다. 다음은 간단한 예입니다.

import { map } from 'rxjs/operators';
 
const nums = of(1, 2, 3);
 
const squareValues = map((val: number) => val * val);
const squaredNums = squareValues(nums);
 
squaredNums.subscribe(x => console.log(x));
 
// Logs
// 1
// 4
// 9

파이프를 사용하여 연산자를 함께 연결할 수 있습니다. 파이프를 사용하면 여러 함수를 단일 함수로 결합 할 수 있습니다. pipe () 함수는 결합 할 함수를 인수로 취하고 실행될 때 순서대로 작성된 함수를 실행하는 새 함수를 리턴합니다.

observable에 적용되는 연산자 세트는 레시피입니다. 즉, 관심있는 값을 생성하기위한 지침 세트입니다. 자체적으로 레시피는 아무 것도하지 않습니다. 레서피를 통해 결과를 얻으려면 subscribe ()를 호출해야합니다.

다음은 그 예입니다.

import { filter, map } from 'rxjs/operators';
 
const nums = of(1, 2, 3, 4, 5);
 
// Create a function that accepts an Observable.
const squareOddVals = pipe(
  filter((n: number) => n % 2 !== 0),
  map(n => n * n)
);
 
// Create an Observable that will run the filter and map functions
const squareOdd = squareOddVals(nums);
 
// Suscribe to run the combined functions
squareOdd.subscribe(x => console.log(x));

pipe() 함수는 RxJS Observable의 메소드이기도하므로이 짧은 형식을 사용하여 동일한 연산을 정의합니다.

import { filter, map } from 'rxjs/operators';
 
const squareOdd = of(1, 2, 3, 4, 5)
  .pipe(
    filter(n => n % 2 !== 0),
    map(n => n * n)
  );
 
// Subscribe to get values
squareOdd.subscribe(x => console.log(x));

RxJS는 많은 연산자를 제공하지만 소수만 자주 사용됩니다. 운영자 및 사용 샘플 목록을 보려면 RxJS API 설명서를 방문하십시오.

앵귤러 앱의 경우 체인을 연결하는 대신 연산자를 파이프와 결합하는 것이 더 좋습니다. 연결은 많은 RxJS 예제에서 사용됩니다.

RxJS는 subscription에 제공하는 error() 처리기 외에도 관찰 가능한 처리법에서 알려진 오류를 처리 할 수있는 catchError 연산자를 제공합니다.

예를 들어, API 요청을 만들고 서버의 응답에 매핑되는 관찰 가능 항목이 있다고 가정합니다. 서버가 오류를 리턴하거나 값이 존재하지 않으면 오류가 발생합니다. 이 오류를 포착하여 기본값을 제공하면 스트림은 오류가 아닌 값을 계속 처리합니다.

다음은 catchError 연산자를 사용하여이를 수행하는 예제입니다.

import { ajax } from 'rxjs/ajax';
import { map, catchError } from 'rxjs/operators';
// Return "response" from the API. If an error happens,
// return an empty array.
const apiData = ajax('/api/data').pipe(
  map(res => {
    if (!res.response) {
      throw new Error('Value expected!');
    }
    return res.response;
  }),
  catchError(err => of([]))
);
 
apiData.subscribe({
  next(x) { console.log('data: ', x); },
  error(err) { console.log('errors already caught... will not run'); }
});

catchError 연산자가 단순 복구 경로를 제공하는 경우, 재시도 연산자를 사용하여 실패한 요청을 재 시도 할 수 있습니다.

catchError 연산자 앞에 retry 연산자를 사용하십시오. 이는 관찰 가능한 원본 소스에 다시 구독하고 오류를 유발 한 모든 동작 시퀀스를 다시 실행할 수 있습니다. 여기에 HTTP 요청이 포함되어 있으면 해당 HTTP 요청을 다시 시도합니다.

다음은 오류를 잡기 전에 이전 예제를 변환하여 요청을 다시 시도합니다.

import { ajax } from 'rxjs/ajax';
import { map, retry, catchError } from 'rxjs/operators';
 
const apiData = ajax('/api/data').pipe(
  retry(3), // Retry up to 3 times before failing
  map(res => {
    if (!res.response) {
      throw new Error('Value expected!');
    }
    return res.response;
  }),
  catchError(err => of([]))
);
 
apiData.subscribe({
  next(x) { console.log('data: ', x); },
  error(err) { console.log('errors already caught... will not run'); }
});

인증(authentication) request는 사용자 작업에 의해서만 시작되어야하므로 다시 시도하지 마십시오. 사용자가 시작하지 않은 반복 된 로그인 요청으로 사용자 계정을 잠그고 싶지 않습니다.

각도 응용 프로그램은 대부분 TypeScript로 작성되기 때문에 일반적으로 변수가 관찰 가능 일 때를 알 수 있습니다. Angular 프레임 워크는 관측 대상에 대한 명명 규칙을 적용하지 않지만 후행하는 “$”기호로 명명 된 관측 대상을 자주 볼 수 있습니다.

코드를 스캔하여 관찰 가능한 값을 찾을 때 유용 할 수 있습니다. 또한, 관측 대상에서 가장 최근 값을 저장하는 속성을 원하면 “$”의 유무와 상관없이 동일한 이름을 사용하는 것이 편리 할 수 있습니다.

예 :

import { Component } from '@angular/core';
import { Observable } from 'rxjs';
 
@Component({
  selector: 'app-stopwatch',
  templateUrl: './stopwatch.component.html'
})
export class StopwatchComponent {
 
  stopwatchValue: number;
  stopwatchValue$: Observable<number>;
 
  start() {
    this.stopwatchValue$.subscribe(num =>
      this.stopwatchValue = num
    );
  }
}