RxJava - Flowable, Observable, Single, Maybe, Completable (4/18)

RxJava

• Netflix 사에서 개발
• Flowable, Observable, Single, Maybe, Completable, publisher 제공

RxJava - Flowable

• 0..n개의 item 전달
• 에러가 발생하면 error signal 전달 후 종료
• 모든 item을 전달 -> complete signal 전달 후 종료
• backPressure 지원
• Reactor의 Flux와 유사

Flowable

@Slf4j
public class FlowableExample {
    public static void main(String[] args) {
        log.info("start main");
        getItems()
                .subscribe(new SimpleSubscriber <>(Integer.MAX_VALUE));
        log.info("end main");
    }
    private static Flowable<Integer> getItems() {
        return Flowable.fromIterable(List.of(1, 2, 3, 4, 5));
    }
}

• backPressure 없이 처리 후 Complete
• 

Flowable - backPressure

@Slf4j
public class FlowableContinuousRequestSubscriberExample {
    public static void main(String[] args) {
        log.info("start main");
        getItems()
                .subscribe(new ContinuousRequestSubscriber <>());		// 1개 처리 + 1개 요청
        log.info("end main");
    }
    private static Flowable<Integer> getItems() {
        return Flowable.fromIterable(List.of(1, 2, 3, 4, 5));
    }
}

• ContinuousRequestSubscriber는 1개 처리 + 1개 요청 Helper class
• 1~5번까지 1개씩 backPressure 적용
• 

RxJava - Observable

• 0..n개의 item 전달
• 에러가 발생하면 error signal 전달 후 종료
• 모든 item을 전달 -> complete signal 전달 후 종료
• backPressure X

Observable vs Flowable

• Push 기반의 차이
  • Observable은 소비자에게 데이터를 push
  • Flowable 또한 push 방식이지만, backPressure 적용
• backPressure 적용 차이
  • Observable은 Subscriber가 처리할 수 없어도, item 전달
  • Flowable은 Subscriber가 request의 수를 조절 가능
• Reactive manifesto
  • Observable은 message driven을 일부만 준수
  • Flowable은 message driven을 모두 준수
• onSubscribe 시
  • Observable은 Disposable 전달 (dispose로 구독 해제 가능)
  • Flowable은 Subscription 전달 (request, cancel)

RxJava - Single

• 1개의 item 전달 후 바로 onComplete signal 전달 (onSuccess)
• 1개의 item이 없다면 onError signal 전달
• 에러가 발생했다면 onError signal 전달
• 에러가 아닌 이상 값이 있는걸 보장
  • ex) count가 0이라고 가정, 'count가 없다' 라고 간주하지 않고, 0이라는 값의 존재로 간주해야함

@Slf4j
public class SingleNullExample {
    public static void main(String[] args) {
        getItem()
                .subscribe(new SimpleSingleObserver <>());
    }
    private static Single<Integer> getItem() {
        return Single.create(singleEmitter -> {
            singleEmitter.onSuccess(null);
        });
    }
} 

// [main] - error: onSuccess called with a null value. Null values are generally not allowed in 3.x operators and sources 에러 발생

RxJava - Maybe

• 1개의 item 전달 후 바로 onComplete signal 전달
• 1개의 item이 없어도 onComplete signal 전달 가능
• 에러가 발생했다면 onError signal 전달
• Mono와 유사
• 값이 있거나 / 없거나 / 에러

RxJava - Completable

• onComplete 혹은 onError signal만 전달
• 값이 아닌 사건을 전달
• 값이 있거나 / 에러

Publisher 요약

• Flowable - 여러개 + backPressure
• Observable - 여러개
• Single - 값 무조건 존재
• Maybe - 값이 없을 수도?
• Completable - 사건