RxJava never: 아무 신호도 오지 않는 스트림

Flowable.never()는 구독자에게 onNext, onError, onComplete 중 어느 것도 보내지 않는다. empty()가 값 없이 즉시 완료하는 것과 다르다. 따라서 출력이 없다는 사실만으로 두 연산자를 구분할 수 없다. 완료 콜백이 호출되는지 살펴봐야 한다.

그림의 수평선에 원(값), 세로선(완료), X(오류)가 모두 없다. 시간이 흘러도 종료 신호가 나타나지 않는다는 뜻이다. 그림은 실제 시간 단위나 내부 작업 스레드를 지정하지 않는다.

구독부터 완료까지 추적하기

import io.reactivex.Flowable;

public class NeverExample {
    public static void main(String[] args) {
        Flowable<String> source = Flowable.never();
        System.out.println("구독 전");
        source.subscribe(
            value -> System.out.println("값: " + value),
            error -> System.out.println("오류: " + error),
            () -> System.out.println("완료")
        );
        System.out.println("구독 후");
    }
}

예상 출력은 구독 전, 구독 후 두 줄뿐이다. never()의 생성과 구독 자체는 즉시 돌아오지만, 구독자에게 알릴 이벤트가 없다. 기본으로 별도 스케줄러를 만들거나 스레드를 잠들게 하는 연산자가 아니다. 그래서 위 콘솔 프로그램은 main이 끝나면 종료된다. 웹 요청에서 응답을 이 스트림의 완료에 연결하면 완료되지 않는 요청이 된다.

조합 연산자 안에 넣으면

Flowable<String> sequence = Flowable.concat(
    Flowable.just("A"),
    Flowable.<String>never(),
    Flowable.just("B")
);
sequence.subscribe(System.out::println);
// A만 출력. 두 번째 소스가 끝나지 않아 B 소스를 구독하지 못함

concat은 첫 소스가 완료한 뒤 다음 소스를 구독한다. A는 통지되고 첫 소스가 완료하지만, 두 번째 never가 끝나지 않으므로 세 번째 B에 도달하지 못한다. 같은 자리에 Flowable.empty()를 넣으면 바로 다음으로 넘어가 A, B가 출력되고 전체도 완료된다. 여러 작업을 연결할 때 중간 작업의 완료 신호가 진행 조건이라는 것을 이 예가 보여 준다.

타임아웃과 취소

Flowable.<String>never()
    .timeout(1, java.util.concurrent.TimeUnit.SECONDS)
    .subscribe(
        System.out::println,
        error -> System.out.println(error.getClass().getSimpleName())
    );
// 구독이 유지되면 약 1초 뒤 TimeoutException을 오류 콜백에 전달

timeout을 붙이면 시간 제한 연산자가 스케줄러로 시간을 감시하고, 기한 내 신호가 없을 때 onError를 보낸다. 이때의 TimeoutException은 원래 never가 보낸 오류가 아니다. 짧은 main은 비동기 타이머 전에 끝날 수 있으므로 테스트에서는 종료를 기다리거나 TestScheduler로 가상 시간을 이동한다.

구독을 해제하면 타임아웃 같은 뒤 작업도 더 이상 필요하지 않다. UI 화면이나 서버 요청의 수명이 끝나면 반환된 구독 핸들을 dispose()하거나 Subscription.cancel()해야 한다. never는 타임아웃·취소·대기 UI의 동작을 시험할 때 유용하지만, 정상적으로 값이 없는 결과를 나타내는 데에는 empty()를 사용한다.

참고: ReactiveX Empty/Never/Error, RxJava 2 Flowable API