실습: Node에서 커스텀 타이머 Observable 직접 구현하기

1. 개념

커스텀 타이머 Observable이란 RxJS가 미리 만들어 둔 timer, interval 오퍼레이터를 쓰지 않고, Observable 생성자를 직접 호출해서 일정 시간마다 값을 방출하는 스트림을 손수 만드는 것을 말한다. RxJS의 Observable은 결국 "구독자(subscriber)가 구독(subscribe)했을 때 어떤 함수를 실행할지"를 정의하는 얇은 래퍼(wrapper)이기 때문에, 그 내부 로직을 Node의 setInterval이나 setTimeout으로 직접 채워 넣을 수 있다.

2. 왜 사용하는가

RxJS를 처음 배울 때 interval(1000) 같은 내장 함수만 사용하면 그 뒤에서 어떤 일이 벌어지는지 감이 잡히지 않는다. 직접 Observable을 만들어보면 구독 함수가 언제 실행되는지, next 호출이 실제로 무엇을 의미하는지, 그리고 구독 해제(unsubscribe) 시 자원을 정리하는 코드가 왜 필요한지를 몸으로 익히게 된다. 이후에 WebSocket, 파일 시스템 감시, 커스텀 이벤트 소스 등을 Observable로 감싸는 실무 작업을 할 때 이 패턴이 그대로 재사용된다.

3. 동작 원리

Observable 생성자는 하나의 함수를 인자로 받는다. 이 함수는 구독자(subscriber) 객체를 매개변수로 받으며, 누군가 subscribe()를 호출하는 순간에만 실행된다. 즉 Observable을 만드는 시점에는 아무 일도 일어나지 않고, 구독이 일어나야 비로소 내부 로직이 동작한다. 이런 특성 때문에 RxJS의 기본 Observable을 콜드(cold) Observable이라고 부르는데, 구독할 때마다 로직이 처음부터 다시 실행된다는 뜻이다.

구독자 객체는 세 가지 메서드를 제공한다. subscriber.next(값)은 새로운 값을 스트림에 흘려보내고, subscriber.complete()는 더 이상 값이 없음을 알리며 스트림을 종료하고, subscriber.error(에러)는 에러가 발생했음을 알린다. 타이머의 경우 setInterval을 이용해서 일정 주기로 next를 호출하면 된다.

마지막으로 구독 함수는 정리(teardown) 함수를 반환할 수 있다. 이 반환된 함수는 구독이 해제(unsubscribe)되거나 스트림이 완료(complete)될 때 자동으로 호출된다. 타이머를 만들 때는 이 자리에 clearInterval을 넣어서, 구독이 끝나면 타이머가 계속 돌아가지 않도록 정리해야 한다.

4. 예제

먼저 rxjs에서 Observable을 가져온다. const { Observable } = require('rxjs');

그다음 Observable 생성자를 사용해서 커스텀 타이머를 만든다. 생성자에는 구독자(subscriber)를 매개변수로 받는 함수를 전달한다. const customTimer = new Observable(subscriber => { ... });

이 함수 내부에서는 방출할 값을 세기 위한 카운터를 하나 만든다. let count = 0;

그리고 setInterval을 이용해서 1초마다 카운터 값을 next로 방출하고, 카운터를 하나씩 증가시킨다. const intervalId = setInterval(() => { subscriber.next(count); count++; }, 1000);

만약 값이 5번 방출된 뒤 스트림을 종료하고 싶다면, next 호출 다음에 조건문을 추가해서 count가 5에 도달하면 subscriber.complete()를 호출하고 clearInterval(intervalId)도 함께 실행한다. if (count >= 5) { subscriber.complete(); clearInterval(intervalId); }

마지막으로 구독 함수는 정리 함수를 반환해야 한다. 구독 해제 시 인터벌을 확실히 제거하기 위해서다. return () => { clearInterval(intervalId); console.log('타이머 정리됨'); };

이제 만든 Observable을 구독해본다. const subscription = customTimer.subscribe({ next: value => console.log('받은 값:', value), complete: () => console.log('완료됨') });

일정 시간 뒤 구독을 강제로 해제하고 싶다면 setTimeout을 이용해서 다음과 같이 호출한다. setTimeout(() => subscription.unsubscribe(), 3500);

이렇게 하면 3.5초 시점에 구독이 해제되면서 앞서 반환했던 정리 함수가 실행되고, 콘솔에 "타이머 정리됨"이 출력되면서 setInterval이 멈춘다.

5. 실무 사용 사례

이 패턴은 RxJS가 기본으로 지원하지 않는 이벤트 소스를 Observable로 감쌀 때 그대로 사용된다. 예를 들어 Node의 EventEmitter 기반 라이브러리, 파일 시스템 변경을 감지하는 fs.watch, 또는 외부 API의 웹훅(webhook) 콜백을 Observable 스트림으로 통합하고 싶을 때 Observable 생성자 안에서 이벤트 리스너를 등록하고, 정리 함수 안에서 리스너를 해제하는 구조를 그대로 응용할 수 있다. 커스텀 타이머 실습은 이런 실무 패턴을 이해하기 위한 가장 단순한 출발점이다.

6. 주의할 점

구독 해제 시 정리 함수를 작성하지 않으면 setInterval이 백그라운드에서 계속 실행되면서 메모리 누수와 불필요한 자원 소모로 이어진다. 구독자 객체에서 next, complete, error를 호출한 뒤에도 코드가 계속 실행될 수 있으므로, complete나 error를 호출한 직후에는 반드시 관련 로직을 멈추도록 처리해야 한다. 또한 이 Observable은 콜드(cold)이기 때문에 여러 곳에서 구독하면 각 구독마다 별도의 setInterval이 새로 생성된다는 점도 기억해야 한다. 여러 구독자가 같은 타이머를 공유하게 하고 싶다면 share나 Subject 같은 별도의 도구가 필요하다.

7. 핵심 정리

커스텀 타이머 Observable 구현은 Observable 생성자에 구독 함수를 전달하고, 그 함수 안에서 next로 값을 방출하고, complete나 error로 스트림을 종료하며, 마지막으로 정리(cleanup) 함수를 반환하는 네 단계로 이루어진다. 이 실습을 통해 Observable이 구독 시점에만 동작하는 콜드 스트림이라는 점과, 자원 정리를 스스로 책임져야 한다는 점을 직접 확인할 수 있다. 이 구조는 이후 RxJS의 내장 오퍼레이터나 외부 이벤트 소스를 Observable로 통합하는 모든 작업의 기초가 된다.