RxJava just: 준비된 값을 순서대로 보내기

Flowable.just("A", "B", "C")는 전달된 값을 순서대로 onNext로 내보내고 마지막에 onComplete를 보낸다. RxJava 2의 just 오버로드에는 인자 수 제한이 있으므로 값이 많거나 배열·컬렉션에 있다면 fromArray·fromIterable을 선택한다. just는 값의 생성 방법이 아니라 이미 준비된 값의 통지 방법이다.

그림에서 원은 통지되는 값, 끝의 세로선은 완료를 뜻한다. A, B, C 사이의 거리가 실제 지연 시간을 뜻하지 않는다. 기본 just는 시간 간격을 넣거나 다른 스레드로 옮기지 않는다.

구독 시 실행 순서

import io.reactivex.Flowable;

public class JustExample {
    public static void main(String[] args) {
        Flowable<String> source = Flowable.just("A", "B", "C");
        System.out.println("구독 전");
        source.subscribe(
            value -> System.out.println(Thread.currentThread().getName() + ": " + value),
            error -> System.err.println(error),
            () -> System.out.println("완료")
        );
        System.out.println("구독 후");
    }
}

별도 스케줄러가 없으면 예상 순서는 구독 전 → main: A → main: B → main: C → 완료 → 구독 후다. source를 만든 줄에서 값을 통지하지 않고 구독할 때 통지한다. 값 세 개를 모두 보낸 뒤 완료하므로 뒤에 .concatWith(Flowable.just("D"))를 붙이면 그 다음에 D가 나온다. 중간에 구독을 취소하는 구독자라면 남은 값과 완료 신호를 받지 않을 수 있다.

값 자체는 언제 계산되는가

int[] calls = {0};
Flowable<Integer> source = Flowable.just(++calls[0]);
System.out.println(calls[0]);       // 1: just의 인자를 먼저 계산
source.subscribe(System.out::println); // 1
source.subscribe(System.out::println); // 다시 1

just의 인자 표현식은 메서드를 호출하기 전 Java가 평가한다. just(loadFromDatabase())라고 쓰면 조회는 구독보다 먼저 일어난다. 구독자마다 값을 새로 계산하려면 fromCallable(() -> loadFromDatabase()) 또는 스트림 생성 자체를 미뤄야 하면 defer를 사용한다. 이미 만들어진 가변 객체를 just에 넘기면 두 구독자가 같은 객체 참조를 보게 된다. 필요하면 불변 스냅샷을 만들어 전달한다.

just(null)은 RxJava 2에서 허용되지 않는다. 값이 없는 정상 결과는 empty(), 오류는 error(...)로 표현한다. Flowable은 배압 요청 계약을 갖지만, 세 값뿐인 이 예제에 별도 스케줄러나 버퍼 정책을 붙일 이유는 없다. 큰 외부 이벤트 소스를 just 하나로 감싸는 것은 생산 속도와 메모리 문제를 해결하지 못한다.

참고: ReactiveX Just, RxJava 2 Flowable API