RxJS Observable / of / from / lastValueFrom / concatMap, toArray

import { concatMap, from, lastValueFrom, mergeMap, of, toArray } from 'rxjs';

// 유저 id로 유저정보를 가져오는 API
const fetchUser = (id: number) => {
  return new Promise((resolve, reject) => {
    if (id === 1) {
      setTimeout(() => {
        resolve({ user: { name: 'tom' } });
      }, 3000);
    } else if (id === 2) {
      setTimeout(() => {
        resolve({ user: { name: 'ruslan' } });
      }, 1500);
    } else {
      reject('유효하지 않은 id');
    }
  });
};

// promise all

const getUserList = async (_userIds: number[]) => {
  const userPromises = [];
  for (const _id of _userIds) {
    userPromises.push(fetchUser(_id));
  }
  // or
  // const userPromises = _userIds.map((id) => fetchUser(id));

  const userData = await Promise.all(userPromises);
  return userData;
};

console.log('promise all run');
getUserList([1, 2]).then((v) => console.log('promise all: ', v));

// rxjs

const rxGetUserList = async (_userIds: number[]) => {
  const source = of(..._userIds);
  const users = await lastValueFrom(
    from(source).pipe(
      concatMap((id) => fetchUser(id)),
      toArray(),
    ),
  );
  return users;
};
console.log('rxjs run');
rxGetUserList([1, 2]).then((v) => console.log('rxjs: ', v));

 

Observable: 데이터스트림을 의미, 데이터 스트림은 시간에따라 연속적으로 발생하는 값 또는 이벤트의 시퀀스 

 

of: RxJS에서 사용되는 오퍼레이터 중 하나로서, 주어진 값들을 순차적으로 방출하는 Obserbable 생성 

 

lastValueFrom: RxJS에서 제공하는 유틸리티함수 중 하나로서, Observable이 마지막값을 방출할때까지 대기하고 해당 값을 Promise로 반환 

 

firstValueForm: Observable이 마지막값을 방출할때까지 대기. 첫번째 값을 받고 즉시 구독을 취소 

 

from: 데이터 소스/구조 (배열, Promise, Iterable, Observable 등)을 Observable로 변환하여 데이터를 방출하는 역활 

 

pipe: 여러 오퍼레이터들을 연결하여 사용히여 Observable을 변환하거나, 필터링하고 새로운 값을 생성하는것에 사용 

  - lodash의 flow와 유사

 

concatMap / mergeMap: RxJS에서 사용되는 오퍼레이터 중 하나로서, Observable을 방출하는 역활을 하면서 각 항목을 콜백함수 인자로 받아 사용

 

  - concatMap: 순차적으로 실행 / 순서를 보장

  - mergeMap: 병렬적으로 실행 / 순서를 미보장

 

toArray: RxJS에서 사용되는 오퍼레이터 중 하나로서 Observable이 방출한 모든 값을 수집하여 하나의 배열로 변환

 

예제에서는 비동기 함수를 핸들링 하는 로직을 rxjs로 변환해봤습니다.

해당 예제에서는 promise.all을 사용하는게 더 적합해보이는것 같기도 같긴합니다.

 

rxjs의 공식문서를 보면 예시가 보통 setInterval로 되어있습니다.

데이터 스트림이 지속적으로 발생하고, 지속적으로 변경하고 처리해야할때 더욱 적합한갓같단 생각이 들었습니다.