미니 프로젝트: Node CLI로 파일 변경을 감지하고 변환 결과를 실시간 출력하는 데이터 스트림 처리기

1. 개념

실시간 데이터 스트림 처리기는 지속해서 들어오는 이벤트를 관찰 가능 객체(Observable)로 표현하고, 파이프라인을 통해 필요한 데이터로 변환하는 프로그램이다.

이번 프로젝트의 흐름은 다음과 같다.

파일 변경 → 이벤트 정제 → 파일 읽기 → 텍스트 변환 → 터미널 출력

Node의 fs.watch()는 콜백 기반 API다. 이를 fromEventPattern()으로 감싸면 파일 변경을 RxJS 스트림으로 다룰 수 있다. 구독(subscribe)이 시작될 때 감시자를 만들고, 구독이 끝날 때 감시자를 닫는 해제(teardown) 동작도 함께 정의한다.

2. 왜 사용하는가

파일 변경 이벤트는 한 번의 입력으로 끝나지 않는다. 편집기가 파일을 저장할 때 여러 이벤트가 연속으로 발생할 수 있고, 파일을 읽는 동안 다음 변경이 들어올 수도 있다. 파일 삭제, 권한 문제, 일시적인 저장 상태처럼 오류도 계속 발생한다.

콜백만 사용하면 중복 억제, 비동기 작업 순서, 오류 복구, 종료 처리가 여러 위치로 흩어진다. RxJS 파이프라인은 이를 선언적인 단계로 모은다.

• debounceTime()으로 짧은 시간에 몰린 이벤트를 하나로 줄인다.
• filter()로 필요한 이벤트만 통과시킨다.
• concatMap()으로 파일 읽기를 순서대로 처리한다.
• switchMap()으로 이전 작업을 버리고 최신 변경만 처리할 수도 있다.
• catchError()로 개별 파일 읽기 실패가 전체 스트림 종료로 이어지지 않게 한다.
• finalize()로 종료 시 정리 작업을 수행한다.

3. 동작 원리

3.1 콜드 관찰 가능 객체와 자원 수명

fromEventPattern()으로 만든 관찰 가능 객체는 구독이 생길 때 이벤트 처리기를 등록한다. 등록 함수에서 fs.watch()를 시작하고 감시자 객체를 반환한다. 제거 함수에서는 반환받은 감시자에 close()를 호출한다.

이 구조에서는 구독 수명과 파일 감시자 수명이 연결된다.

구독 시작 → fs.watch() 시작 → 이벤트 전달 → 구독 해제 → watcher.close()

3.2 이벤트 폭주 제어

파일 저장 한 번이 여러 변경 이벤트를 만들 수 있다. debounceTime(100)은 마지막 이벤트가 발생한 뒤 100밀리초 동안 새 이벤트가 없을 때만 값을 내보낸다.

같은 파일명이 반복될 때만 억제하려면 distinctUntilChanged()를 함께 사용할 수 있다. 다만 이벤트 값이 항상 같은 문자열이라면 실제로 다른 저장까지 막을 수 있으므로 시간 정보나 경로를 포함한 객체를 비교해야 한다.

3.3 비동기 파일 읽기 평탄화

readFile()은 프라미스(Promise)를 반환한다. from(readFile(...))로 관찰 가능 객체로 바꾼 뒤 고차 매핑 연산자(higher-order mapping operator)로 연결한다.

• concatMap()은 변경을 순서대로 처리한다. 모든 변경 결과가 필요할 때 적합하다.
• switchMap()은 새 변경이 들어오면 이전 내부 구독을 해제하고 최신 작업 결과만 사용한다.
• mergeMap()은 여러 읽기를 동시에 처리한다. 결과 순서는 보장하지 않는다.
• exhaustMap()은 읽는 동안 들어온 변경을 무시한다.

프라미스 기반 파일 읽기는 시작된 운영체제 작업 자체가 취소되지 않을 수 있다. switchMap()은 이전 결과 전달을 막지만, 이미 시작된 readFile() 호출까지 중단한다고 가정하면 안 된다.

3.4 오류 격리

바깥 파이프라인에서 catchError()를 사용하면 오류 처리 뒤 스트림이 완료될 수 있다. 파일 읽기 하나의 실패만 복구하려면 concatMap() 또는 switchMap() 안쪽에서 catchError()를 적용한다.

이렇게 하면 특정 변경 처리에 실패해도 파일 감시는 계속된다.

4. 예제 (인라인 코드로만, fenced 코드블록 금지)

먼저 프로젝트를 만들고 npm init -y를 실행한 뒤 npm install rxjs로 RxJS를 설치한다. package.json에는 "type": "module"을 추가한다.

watch-cli.js에서 import { watch } from 'node:fs', import { readFile } from 'node:fs/promises', import process from 'node:process'를 선언한다. RxJS에서는 import { EMPTY, from, fromEvent, fromEventPattern, of } from 'rxjs'를 가져오고, 연산자는 import { catchError, concatMap, debounceTime, filter, finalize, map, takeUntil, tap } from 'rxjs/operators'로 가져온다.

감시할 경로는 const targetPath = process.argv[2]로 받는다. 인수가 없으면 console.error('사용법: node watch-cli.js <파일 경로>')를 실행하고 process.exit(1)로 종료한다.

파일 감시 스트림은 const changes$ = fromEventPattern(...) 형태로 만든다. 등록 함수에서는 const watcher = watch(targetPath, handler)를 실행하고 return watcher로 감시자를 반환한다. 제거 함수에서는 watcher.close()를 호출한다.

fs.watch()의 콜백은 이벤트 종류와 파일명을 따로 전달하므로 등록 함수 안에서 watch(targetPath, (eventType, filename) => handler({ eventType, filename, detectedAt: Date.now() }))처럼 하나의 객체로 묶는다.

종료 신호는 const stop$ = fromEvent(process, 'SIGINT')로 만든다. 이후 changes$.pipe(...) 안에서 먼저 filter(({ eventType }) => eventType === 'change')로 변경 이벤트만 남기고, 그다음 debounceTime(100)으로 연속 이벤트를 줄인다.

파일 읽기는 concatMap(event => from(readFile(targetPath, 'utf8')).pipe(...))로 연결한다. 내부 파이프라인에서 map(content => ({ event, content }))로 이벤트와 내용을 묶고, catchError(error => { console.error(\읽기 실패: ${error.message}\); return EMPTY; })로 개별 실패를 처리한다.

텍스트 변환은 다음 map()에서 수행한다. 예를 들어 빈 줄을 제거하고 각 줄 앞에 번호를 붙이려면 map(({ event, content }) => ({ event, output: content.split(/\r?\n/).filter(line => line.trim() !== '').map((line, index) => \${index + 1}: ${line.toUpperCase()}\).join('\n') }))를 사용한다.

출력 단계는 tap(({ event, output }) => { console.clear(); console.log(\감지 시각: ${new Date(event.detectedAt).toLocaleTimeString()}\); console.log(output); })로 구성한다.

종료 처리는 takeUntil(stop$)을 연결한 뒤 finalize(() => console.log('\n파일 감시를 종료했습니다.'))를 추가한다. 마지막으로 subscribe()를 호출하면 감시가 시작된다.

실행 명령은 node watch-cli.js ./input.txt다. input.txt를 수정하고 저장하면 빈 줄이 제거되고, 각 줄이 대문자와 줄 번호를 포함한 형태로 터미널에 출력된다. 종료하려면 Ctrl+C를 누른다.

5. 실무 사용 사례

로그 파일 후처리

추가되는 로그를 읽어 로그 수준별로 필터링하고 구조화된 JSON으로 변환할 수 있다. 처리량이 많다면 파일 전체를 다시 읽지 말고 마지막 읽기 위치(offset)를 저장해 새로 추가된 구간만 읽어야 한다.

설정 파일 자동 반영

설정 파일 변경을 감지한 뒤 파싱하고 유효성을 검사해 실행 중인 서비스에 반영할 수 있다. 잘못된 설정은 catchError()로 격리하고 마지막 정상 설정을 유지한다.

정적 파일 변환

마크다운, 스타일시트, 템플릿 파일의 변경을 감지해 빌드 결과를 갱신할 수 있다. 최신 결과만 필요하면 switchMap()이 적합하다.

데이터 수집 파이프라인

특정 디렉터리에 생성되는 CSV나 JSON 파일을 감지하고, 파싱·검증·정규화한 뒤 데이터베이스나 메시지 큐로 전달할 수 있다. 처리 순서가 중요하면 concatMap(), 병렬 처리가 가능하면 동시성 제한을 둔 mergeMap()을 사용한다.

6. 주의할 점

fs.watch() 동작은 운영체제마다 다르다

이벤트 종류, 파일명 제공 여부, 중복 발생 방식이 운영체제와 파일 시스템에 따라 달라질 수 있다. 네트워크 파일 시스템이나 컨테이너 마운트에서도 동작 차이가 생길 수 있다. 이벤트 한 건을 파일 변경 한 번과 동일하게 취급하지 말고, 디바운스(debounce) 후 실제 파일 상태를 다시 읽는 방식이 안전하다.

파일 교체 저장을 고려해야 한다

일부 편집기는 기존 파일을 직접 수정하지 않고 임시 파일을 만든 뒤 원본과 교체한다. 이 경우 기존 파일에 연결된 감시자가 더 이상 변경을 받지 못할 수 있다. 장시간 실행하는 도구라면 파일 자체보다 상위 디렉터리를 감시하고 대상 파일명을 필터링하는 구성이 더 안정적이다.

역압력(backpressure)을 연산자로 설계해야 한다

변경 속도가 처리 속도보다 빠르면 대기 작업이 계속 쌓일 수 있다. concatMap()은 순서를 보장하지만 큐가 길어질 수 있다. 최신 상태만 중요하면 switchMap(), 처리 중 발생한 변경을 버려도 되면 exhaustMap(), 제한된 병렬 처리가 필요하면 동시성 값을 지정한 mergeMap()을 선택한다.

오류 처리 위치가 스트림 수명을 결정한다

개별 작업 오류를 바깥쪽 catchError()에서 처리하면 파일 감시 전체가 끝날 수 있다. 계속 실행해야 하는 CLI에서는 오류가 발생할 수 있는 내부 관찰 가능 객체에 catchError()를 배치한다.

종료 처리를 생략하면 자원이 남는다

감시자, 파일 핸들, 타이머, 하위 프로세스는 구독 해제 시 정리돼야 한다. fromEventPattern()의 제거 함수, takeUntil(), finalize()를 사용해 종료 경로를 명시한다. SIGINT 외에 정상 종료 정책에 따라 SIGTERM도 처리할 수 있다.

7. 핵심 정리

• Node의 콜백 이벤트는 fromEventPattern()으로 관찰 가능 객체(Observable)로 감쌀 수 있다.
• 파일 변경 이벤트는 중복될 수 있으므로 filter()와 debounceTime()으로 정제한다.
• 비동기 작업의 처리 정책에 따라 concatMap(), switchMap(), mergeMap(), exhaustMap()을 선택한다.
• 개별 작업의 catchError()는 오류를 격리해 전체 감시 스트림을 유지한다.
• takeUntil()과 해제(teardown) 함수로 CLI 종료 시 파일 감시 자원을 정리한다.