take, first, takeWhile, takeUntil: 스트림을 원하는 지점에서 끊어내는 완료(complete) 제어 연산자

1. 개념

take(n)은 소스 옵저버블에서 처음 n개의 값만 방출하고 자동으로 완료(complete)시키는 연산자다.

first()는 조건 없이 사용하면 첫 번째 값 하나만 방출하고 완료되며, 조건자(predicate) 함수를 인자로 넘기면 그 조건을 만족하는 첫 번째 값을 방출한다. take(1)과 비슷해 보이지만 값이 하나도 방출되지 않고 소스가 완료되면 에러를 던진다는 차이가 있다.

takeWhile(predicate)은 조건자 함수가 참(true)을 반환하는 동안에만 값을 방출하고, 조건자가 거짓(false)을 반환하는 순간 스트림을 완료시킨다. 이때 거짓을 만든 값 자체는 기본적으로 방출되지 않는다.

takeUntil(notifier)은 소스 옵저버블의 값 자체가 아니라 별도로 넘겨준 notifier 옵저버블이 값을 방출하는 시점을 기준으로 스트림을 완료시킨다. 즉 종료 조건이 스트림 내부 값이 아니라 외부 신호다.

2. 왜 사용하는가

리액티브 스트림은 명시적으로 끊지 않으면 계속 살아있는 구독(subscription) 상태로 남는다. 클릭 이벤트, 웹소켓 메시지, 타이머처럼 무한히 값을 낼 수 있는 소스를 다룰 때, "몇 개만 받겠다", "이 조건까지만 받겠다", "이 시점이 되면 그만 받겠다"라는 종료 규칙을 선언적으로 표현할 방법이 필요하다.

이 네 연산자를 쓰지 않으면 구독 시점에 콜백 안에서 카운터를 세고 조건을 검사한 뒤 수동으로 unsubscribe를 호출하는 명령형 코드를 작성해야 한다. take 계열 연산자는 이런 반복적인 종료 로직을 파이프라인 안으로 끌어들여 코드를 선언적으로 만든다.

3. 동작 원리

네 연산자는 모두 소스 옵저버블을 구독한 뒤, 내부 조건이 충족되면 자기 자신의 구독자에게 complete()를 호출하고 동시에 소스에 대한 구독을 해제(unsubscribe)한다.

take(n)은 내부 카운터를 하나 두고, 값이 방출될 때마다 카운터를 증가시킨다. 카운터가 n에 도달하면 그 값을 방출한 직후 완료 신호를 보내고 소스 구독을 끊는다. n이 0이면 아무 값도 받지 않고 즉시 완료된다.

first()는 값이 하나 들어오면(또는 조건자를 만족하는 값이 들어오면) 그 값을 방출하고 즉시 완료 처리한다. 소스가 값을 하나도 내지 않고 완료(complete)되어 버리면 EmptyError를 던진다는 점에서 take(1)과 실행 결과가 갈린다.

takeWhile(predicate)은 값이 들어올 때마다 predicate(value)를 호출한다. 결과가 true면 값을 그대로 통과시키고, false가 나오는 순간 통과시키지 않고 완료 신호를 보낸다. 두 번째 인자로 inclusive 옵션을 true로 주면 조건을 깨뜨린 마지막 값도 함께 방출한 뒤 완료시킬 수 있다.

takeUntil(notifier)은 소스와 notifier 두 옵저버블을 동시에 구독한다. notifier가 첫 값을 방출하는 순간(에러나 완료 여부와 무관하게 값 방출 자체가 트리거) 소스 구독과 notifier 구독을 모두 해제하고 완료 신호를 보낸다. notifier가 아무것도 방출하지 않으면 소스는 원래대로 동작한다.

4. 예제

take(n)으로 처음 3개만 받는 경우를 보면, interval(1000).pipe(take(3)).subscribe(value => console.log(value))처럼 작성한다. 이렇게 하면 0, 1, 2가 1초 간격으로 출력된 뒤 스트림이 자동으로 완료된다.

first()로 조건에 맞는 첫 값을 받는 경우는, 먼저 from([1, 3, 5, 8, 9])로 숫자 스트림을 만들고, 그다음 .pipe(first(value => value % 2 === 0))를 붙여서 짝수인 첫 값을 찾는다. 이 예제에서는 8이 방출되고 즉시 완료된다.

takeWhile(predicate)로 조건이 깨질 때까지만 받는 경우는, from([1, 2, 3, 4, 1, 2])에서 .pipe(takeWhile(value => value < 4))를 적용하면 1, 2, 3까지만 방출되고 4를 만나는 순간 완료된다. 이때 두 번째 인자를 true로 주면 takeWhile(value => value < 4, true)처럼 작성해서 조건을 깨뜨린 4까지 포함해서 방출할 수 있다.

takeUntil(notifier)로 외부 신호에 따라 종료하는 경우는, 먼저 fromEvent(document, 'click')으로 클릭을 감지하는 notifier를 만들고, 그다음 interval(1000).pipe(takeUntil(clickStream)).subscribe(value => console.log(value))처럼 작성한다. 이렇게 하면 사용자가 화면을 클릭하기 전까지는 1초마다 값이 출력되고, 클릭하는 순간 스트림이 완료된다.

5. 실무 사용 사례

take(n)은 무한 스트림에서 초기 데이터 몇 개만 미리보기(preview)로 보여주거나, 테스트 코드에서 무한 옵저버블의 값을 유한하게 잘라서 검증할 때 자주 쓴다.

first()는 여러 이벤트 소스 중 조건을 만족하는 첫 이벤트만 처리하고 나머지는 무시해야 할 때, 예를 들어 여러 서버 응답 중 성공 상태 코드를 가진 첫 응답만 골라낼 때 사용한다.

takeWhile(predicate)은 로딩 진행률이 100%에 도달할 때까지만 진행 상태를 구독하거나, 특정 값 범위 안에 있는 센서 데이터만 수집하고 범위를 벗어나면 자동으로 스트림을 종료하는 상황에 적합하다.

takeUntil(notifier)은 컴포넌트가 언마운트(unmount)되거나 사용자가 취소 버튼을 누를 때 진행 중인 모든 구독을 한 번에 정리하는 패턴에서 가장 널리 쓰인다. 컴포넌트별로 destroy$라는 Subject를 만들어두고 모든 구독 파이프라인 끝에 takeUntil(this.destroy$)를 붙인 뒤, 정리 시점에 destroy$.next()를 호출하는 방식이 대표적이다.

6. 주의할 점

first()는 소스가 빈 상태로 완료되면 에러를 던지므로, 에러가 발생해도 괜찮은 상황인지 미리 확인해야 한다. 값이 없을 수도 있는 상황이라면 기본값을 지정할 수 있는 별도 옵션이나 defaultIfEmpty와 같은 연산자를 함께 고려해야 한다.

takeWhile(predicate)은 기본적으로 조건을 깨뜨린 값을 방출하지 않으므로, 그 값까지 포함해야 하는 요구사항이 있다면 반드시 inclusive 인자를 true로 지정해야 한다. 이를 놓치면 마지막 유효 값이 누락되는 버그가 생긴다.

takeUntil(notifier)은 파이프라인 안에서의 위치가 중요하다. 다른 연산자보다 뒤에 놓으면 그 앞의 연산자들이 여전히 동작한 뒤에야 종료되므로, 일반적으로는 파이프라인의 가장 마지막 근처에 배치해서 전체 체인을 확실히 끊어주는 것이 안전하다.

take(n)에서 n이 소스가 실제로 방출할 값의 개수보다 크면, take는 소스가 스스로 완료될 때까지 기다렸다가 소스의 완료 신호를 그대로 전달한다. 즉 n을 채우지 못했다고 에러가 나지는 않는다.

7. 핵심 정리

네 연산자는 모두 "언제 스트림을 끊을 것인가"를 선언적으로 표현하는 도구이며, 종료 판단 기준이 각각 다르다. take는 개수, first는 조건에 맞는 첫 값의 존재 여부, takeWhile은 값 자체에 대한 조건, takeUntil은 외부 신호가 기준이다. 실무에서는 컴포넌트 생명주기나 취소 로직과 결합할 때 takeUntil이 가장 널리 쓰이고, 나머지 세 연산자는 데이터 흐름 자체의 조건에 따라 선택적으로 사용한다.