RxJava - What's different in 3.0

RxJava 란 비동기 및 이벤트 기반 프로그래밍을 가능하게 해주는 라이브러리

이제 2.x 버전은 유지보수 모드로 버그 픽스만 적용될 뿐 새로운 기능 업데이트는 없으며,
2021년 3월부터는 모든 개발 중단!

참고) https://github.com/ReactiveX/RxJava/wiki/What's-different-in-3.0

 

1. Gradle import

dependencies { 
    implementation 'io.reactivex.rxjava3:rxandroid:3.0.0' // https://github.com/ReactiveX/RxAndroid
    implementation 'io.reactivex.rxjava3:rxjava:3.0.0'    // https://github.com/ReactiveX/RxJava/releases
    implementation 'io.reactivex.rxjava3:rxkotlin:3.0.0'  // https://github.com/ReactiveX/RxKotlin/releases
}

 

2. Java 8

Android Studio 3.0 이상은 Java 7 기능을 모두 지원하고, Java 8 의 경우 플랫폼 버전에 따라 부분 지원하고 있다.
desugaring 이라는 프로세스 덕분에 특정 Java 8 기능과 3rd-party 라이브러리 지원이 가능해져서
RxJava 의 기준을 Java 8 로 업데이트하고 공식 지원한다.
참고) Use Java 8 language features and APIs

• Stream Collector : standard transformations 으로 지정된 컬렉션으로 아이템을 합친다
• Optional : RxJava 의 요구사항인 non-null 을 지원한다
• CompletableFuture : CompletableFuture 을 non-blocking 하게 사용하거나 CompletableFuture 의 단일 결과를 나타낸다
• non-null 어노테이션 사용 : 특정 상황에서 functional 타입이 null 을 리턴할 수 있게한다.

지원하지 않는 기능 : java.time.Duration / java.util.function

Android Java 8 지원 활성화 설정

android {
    compileOptions {
        sourceCompatibility JavaVersion.VERSION_1_8
        targetCompatibility JavaVersion.VERSION_1_8
    }
}

 

3. Package structure

• RxJava 1 : rx
• RxJava 2 : io.reativex
• RxJava 3 : io.reactivex.rxjava3
• RxJava 의 핵심 타입인 Flowable, Observer 등은 io.reativex.rxjava3.core 로 이동

RxJava 3 에서는 Java 8 런타임이 필요하므로 io.reactivex.rxjava3.functions.Function 대신에
java.util.function.Function 과 같은 표준 라이브러리 기능을 선택할 수 있다.

4. Behavior changes

참고) https://www.charlezz.com/?p=43954

5. API changes

5-1. New Types

Supplier

• RxJava 2.x : call() 메소드가 선언되어 기본 Exception 을 던지는 java.util.concurrent.Callable 지원
• RxJava 3.x : get() 메소드가 선언되어 Exception 을 던지는 io.reactivex.rxjava3.functions.Supplier 도입!

/**
 * 결과를 리턴하고 예외가 발생할 수 있는 구현을 위한 목적이며,
 * Runnable 과 유사하게 다른 스레드에 의해 잠재적으로 실행될 수 있는 
 * 클래스를 위해 설계된 인터페이스
*/
@FunctionalInterface
public interface Callable<V> {
    /**
     * Computes a result, or throws an exception if unable to do so.
     *
     * @return computed result
     * @throws Exception if unable to compute a result
     */
    V call() throws Exception;
}


/**
 * 단일 값을 제공하거나 예외를 발생시키는 인터페이스(callback)
 * Callable 인터페이스로는 Throwables 의 subclass 를 던질 수 있도록 추가됨
*/
@FunctionalInterface
public interface Supplier<@NonNull T> {

    /**
     * Produces a value or throws an exception.
     * @return the value produced
     * @throws Throwable if the implementation wishes to throw any type of exception
     */
    T get() throws Throwable;
}

Converters

• to : Observable 을 다른 객체 또는 데이터 구조로 변환
• XConverter
  • Observable.to() 연산자가 Observable 을 다른 값으로 잘 전환하기 위해 사용하는 편의 인터페이스 (callback)
  • 업스트림 Observable 에 함수를 적용하고 R 타입으로 리턴

// before
source.to(new Function<Flowable<Integer>, Integer>() {
    @Override
    public Integer apply(Flowable<Integer> t) throws Exception {
        return t.blockingFirst();
    }
});


// after
source.to(new FlowableConverter<Integer, Integer>() {
    @Override
    public Integer apply(Flowable<Integer> t) {
        return t.blockingFirst();
    }
});


@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
@NonNull
public final <R> R to(@NonNull ObservableConverter<T, ? extends R> converter) {
    return Objects.requireNonNull(converter, "converter is null").apply(this);
}  

 

5-2. Moved components

Disposables & DisposableContainer

• Java 8 및 Android 의 desugaring 도구로 별도의 팩토리 클래스 대신 정적 인터페이스 메소드 사용
• 패키지가 io.reativex -> io.reactivex.java3 로 이동

 

5-3. API Promotion

2.2.x 의 실험적인 연산자들이 3.x 에도 포함

• Flowable  : dematerialize(Function)
• Observable  : dematerialize(Function)
• Single : dematerialize(Function), materialize()
• Maybe : doOnTerminate(Action), materialize()
• Completable : delaySubscription(long, TimeUnit, Scheduler), materialize()

materialize() & dematerialize()

• Observable 은 observer 의 onNext 를 0번 이상 호출한 다음 onCompleted/onError 메소드를 한 번만 호출함
• materialize()
  : 위 호출들을 Observable 이 생성한 일련의 아이템으로 변환
• dematerialize()
  : 위 과정을 반대로 수행. materialize() 에 의해 변환된 Observable 에서 동작해서 원래 형태로 되돌림

doOnTerminate()

• Observable 이 error, complete 를 발행하고 종료하기 직전에 호출 될 Action 을 등록

delaySubscription

• Flowable/Observable/Completable 의 처리 시작을 지연하는 연산자
• 지정한 시간만큼 처리 시작 시간을 지연하고 아이템 생성하면 바로 알림
  
  

5-4. API additions

• 너무 많음,, 문서 참고
  
  

6. Java 8 additions

• API 기준이 Java 8 로 설정되어 이제 외부 라이브러리(ex. RxJavaJdk8Interop) 없이 새로운 타입의 Java 8 직접 지원 가능
• Java 8 의 functional interface 를 직접 지원하지는 않음
  더 많은 플랫폼 지원 및 쉬운 에러 핸들링을 위해 RxJava 에 포함된 functional interface 를 선호한다!
• Java 8 패키지에 포함된 타임을 Flowable, Observable 등으로 변환을 지원한다
• 너무 많아서 문서 참고
  
  

7. API renames

•  onErrorResumeNext -> onErrorResumeWith
  : 이름이 애매하고 다른 언어에서 잘못된 사용이 있어서 변경함
• 나머지는 문서 참고
  
  

8. API signature changes

• Callable -> Supplier : 5-1 참고
• 나머지는 문서 참고,,
  
  

9. API removals

dematerialize

• Flowable/Observable 의 dematerialize(Function) 는 형식이 안전하지 않아서 제거되었다는데,
  위 5-3 의 실험적인 API 에는 아직 있음,, 뭐지?
• 대신 deserialize(Function) 을 사용하라고 한다

zip

• 알려진 source 의 수가 필요하다고 함
• Flowable/Observable 에서 제거되었고 아래처럼 에뮬레이트 할 수 있다함

Observable<Observable<Integer>> sources = ...

sources.toList().flatMapObservable(list -> Observable.zip(list, zipper));

 

나머지는 문서 참고,,

10. Interoeration

• 2.x 에서 3.x 로의 마이그레이션을 돕기위해 외부 interop 라이브러리 제공