pipe()로 연산자 연결하는 패턴: 여러 연산자(operator)를 하나의 파이프라인으로 조립하는 방법

1. 개념

pipe()는 옵저버블 인스턴스에 딸린 메서드로, 여러 연산자를 인자로 받아 순서대로 적용한 뒤 최종 옵저버블을 반환한다. 예를 들어 source$.pipe(map(fn1), filter(fn2), tap(fn3))라고 쓰면, source$가 방출하는 값이 map → filter → tap 순서로 흘러가며 각 단계에서 변형되거나 걸러진다.

여기서 연산자(operator)란 옵저버블을 인자로 받아 새로운 옵저버블을 반환하는 함수를 말한다. map(fn1) 자체는 옵저버블을 받아 옵저버블을 돌려주는 함수이고, pipe()는 이런 함수들을 순서대로 실행해주는 조립 도구일 뿐이다.

2. 왜 사용하는가

RxJS 초기 버전에서는 연산자를 메서드 체이닝(chaining) 방식으로 사용했다. 예를 들어 source$.map(fn1).filter(fn2)처럼 옵저버블 프로토타입에 직접 메서드를 추가하는 방식이었다. 이 방식은 사용하지 않는 연산자까지도 전부 임포트(import)해서 프로토타입을 오염시켜야 했기 때문에, 트리 셰이킹(tree-shaking)이 되지 않아 번들 크기가 커지는 문제가 있었다.

pipe() 방식은 연산자를 독립된 순수 함수로 분리했다. 필요한 연산자만 개별적으로 임포트해서 pipe() 안에 나열하면 되므로, 사용하지 않는 코드는 번들에서 제외된다. 또한 연산자가 옵저버블 프로토타입과 분리되어 있어서 커스텀 연산자를 만들거나 테스트하기도 쉽다.

3. 동작 원리

pipe()는 내부적으로 함수 합성(function composition)을 수행한다. source$.pipe(opA, opB, opC)는 개념적으로 opC(opB(opA(source$)))와 같다. 각 연산자는 "옵저버블을 받아서 옵저버블을 반환하는 함수"이므로, 앞 연산자의 반환값이 다음 연산자의 입력값으로 그대로 전달된다.

구독(subscribe)이 일어나기 전까지는 이 파이프라인이 실제로 값을 흘려보내지 않는다. pipe()는 옵저버블을 조립하는 단계이고, 실제 데이터 방출과 변형은 누군가 .subscribe()를 호출한 시점부터 시작된다. 이는 옵저버블이 지연 실행(lazy evaluation) 방식으로 동작한다는 특성과 연결된다.

4. 예제

먼저 of(1, 2, 3, 4, 5)로 숫자를 방출하는 옵저버블을 만들고, 여기에 pipe()를 연결해서 짝수만 남기고 각 값을 제곱한 뒤 콘솔에 로그를 남기는 흐름을 만들어본다.

source$.pipe(filter(n => n % 2 === 0), map(n => n * n), tap(n => console.log('중간값:', n)))라고 작성하면, 먼저 filter가 홀수를 걸러내고, 그다음 map이 남은 짝수를 제곱하고, 마지막으로 tap이 부수 효과(side effect)로 로그를 출력한다.

이 파이프라인은 아직 아무 값도 흘리지 않은 상태이므로, 그다음 .subscribe(result => console.log('최종값:', result))를 호출해야 실제로 2와 4가 걸러져서 4와 16으로 변환되고 로그가 출력된다.

연산자를 여러 단계로 나눠서 읽기 쉽게 만들 수도 있다. 예를 들어 debounceTime(300)으로 입력 이벤트를 지연시키고, 그다음 distinctUntilChanged()로 같은 값이 연속되면 무시하고, 그다음 switchMap(query => ajaxFetch(query))로 검색 요청을 보내는 식으로, 각 연산자가 하나의 명확한 역할만 담당하도록 나누면 파이프라인 전체의 의도를 읽기 쉬워진다.

5. 실무 사용 사례

검색어 입력창에서 타이핑이 끝난 뒤 API를 호출하는 자동완성(autocomplete) 기능을 만들 때, debounceTime으로 타이핑 완료를 기다리고, distinctUntilChanged로 중복 요청을 막고, switchMap으로 이전 요청을 취소하며 최신 요청만 유지하는 패턴을 pipe() 하나로 조립한다.

폼(form) 값 검증에서는 map으로 입력값을 정규화하고, filter로 유효하지 않은 값을 걸러내고, tap으로 로깅이나 UI 업데이트 같은 부수 효과를 처리하는 구조를 자주 쓴다.

에러 처리도 파이프라인 안에서 조립한다. catchError를 다른 변환 연산자 뒤에 배치해서, 앞 단계에서 발생한 에러를 잡아 대체 값을 방출하거나 스트림을 재시도(retry)하는 흐름을 만든다.

6. 주의할 점

연산자의 순서가 결과를 바꾼다. filter를 먼저 적용한 뒤 map을 적용하는 것과, map을 먼저 적용한 뒤 filter를 적용하는 것은 서로 다른 결과를 낼 수 있다. 예를 들어 짝수만 거른 뒤 제곱하는 것과, 먼저 제곱한 뒤 짝수만 거르는 것은 필터링 조건이 적용되는 시점이 다르므로 결과 집합이 달라질 수 있다.

tap은 부수 효과를 위한 연산자이므로 스트림의 값을 변형하는 용도로 쓰면 안 된다. 로깅이나 디버깅처럼 값 자체를 바꾸지 않는 작업에만 사용하는 것이 원칙이다.

한 파이프라인 안에 너무 많은 연산자를 나열하면 가독성이 떨어진다. 이런 경우 관련 연산자들을 묶어 별도의 함수로 분리하고, 그 함수를 pipe() 안에 인자로 넣는 방식으로 재사용성과 가독성을 높일 수 있다.

switchMap, mergeMap, concatMap처럼 내부에서 새로운 옵저버블을 생성하는 고차(higher-order) 연산자를 잘못된 위치에 배치하면 요청이 취소되지 않거나 중복 실행되는 문제가 생길 수 있으므로, 이런 연산자를 쓸 때는 앞뒤에 어떤 연산자가 오는지 신경 써야 한다.

7. 핵심 정리

pipe()는 여러 연산자를 좌에서 우로 함수 합성해 하나의 옵저버블 파이프라인을 만드는 도구다. 각 연산자는 옵저버블을 받아 옵저버블을 반환하는 순수 함수이며, 트리 셰이킹과 재사용성을 위해 메서드 체이닝 대신 도입되었다. 파이프라인은 구독 전까지 실행되지 않으며, 연산자의 순서와 역할 분리가 결과와 가독성에 직접 영향을 준다.