RxJava fromArray와 fromIterable

배열이나 컬렉션에 여러 값이 있을 때 fromArray와 fromIterable은 원소를 하나씩 onNext로 내보낸다. 컬렉션 전체를 한 번에 통지하는 것과 다르다. fromArray는 배열을, fromIterable은 순회할 수 있는 Iterable을 받는다. 둘 다 값을 끝까지 전달하면 onComplete로 종료한다.

그림의 연속된 원은 배열·목록의 각 원소다. 마지막 완료선은 순회가 끝났음을 나타낸다. 이 다이어그램에는 자료구조를 만드는 시간이나 별도 스레드가 표현되지 않는다.

배열과 목록의 실행 흐름

import io.reactivex.Flowable;
import java.util.Arrays;
import java.util.List;

public class FromCollections {
    public static void main(String[] args) {
        String[] array = {"A", "B", "C"};
        List<String> list = Arrays.asList(array);

        Flowable.fromArray(array).subscribe(System.out::println);
        Flowable.fromIterable(list).subscribe(System.out::println);
        // 각각 A, B, C를 순서대로 통지한 뒤 완료
    }
}

별도 스케줄러가 없으므로 두 구독은 main 스레드에서 차례로 실행된다. fromArray(array)가 배열 원소를 통지한다는 점을 분명히 하려면 배열 변수를 넘기는 이 형태를 쓴다. fromIterable은 구독할 때 iterator()를 얻어 순회한다. Arrays.asList(array)는 원본 배열을 바탕으로 한 고정 크기 목록이므로 배열 원소를 바꾸면 목록에 보이는 값도 달라진다. 구독 시점의 고정된 데이터를 기대한다면 먼저 복사본을 만든다.

List<String> snapshot = List.copyOf(list);
Flowable<String> source = Flowable.fromIterable(snapshot);
source.subscribe(System.out::println); // A, B, C

List.copyOf는 이 예제의 문자열 값들을 목록에 담은 불변 스냅샷을 만든다. 원소 자체가 가변 객체라면 깊은 복사는 아니다. 소스를 만든 뒤 구독하기 전에 원본 컬렉션을 수정하거나 순회 중에 다른 스레드가 수정하면 결과가 달라지거나 예외가 발생할 수 있다. 원본의 수명과 변경 가능성을 구독 시점까지 고려한다.

끝까지 읽지 않는 구독

Flowable.fromIterable(List.of("A", "B", "C", "D"))
    .take(2)
    .subscribe(
        System.out::println,
        Throwable::printStackTrace,
        () -> System.out.println("완료")
    );
// A, B, 완료

take(2)는 두 값을 받은 뒤 상류 구독을 취소하고 결과 스트림을 완료한다. 원본 목록에 C, D가 남아 있어도 결과에 나오지 않는다. 느리게 값을 생산하는 Iterable이라면 조기 종료로 불필요한 순회를 줄일 수 있다. 하지만 이미 메모리에 만들어 둔 큰 목록의 메모리 사용을 없애지는 못한다.

null 원소가 들어 있는 배열·목록은 RxJava 2의 값 계약과 맞지 않아 통지 중 오류가 난다. 값을 건너뛸지, 오류로 처리할지 원본 자료를 검사해 결정한다. 네트워크 이벤트처럼 끝이 없고 생산 속도가 빠른 자료는 단순 컬렉션 연산자보다 수명과 배압을 명시한 소스가 필요하다.

참고: ReactiveX From, RxJava 2 Flowable API