RxJava 안드로이드 프로젝트에 적용하기 01 (with Sunflower)

RxJava 안드로이드 프로젝트에 적용하기 (with Sunflower)

RxJava에 관한 책을 정리한 후 안드로이드 프로젝트에 적용하여 학습하는 방법으로, Google의 Android Sunflower 프로젝트 내부의 코드를 Rx로 migration하는것을 목표로 한다.

Coroutine Flow -> RxJava

Rx는 Single, Maybe, Completable 등 활용할 수 있는 다양한 클래스가 존재한다.

Single

• 데이터를 1건만 통지하거나 에러를 통지
• 데이터 통지가 곧 완료 통지이므로 별도의 완료 통지 안함
• onNext와 onComplete 대신 onSuccess 제공

Maybe

• 데이터를 1건만 통지하거나, 1건도 통지않거나 에러 통지
• 데이터가 1건도 없이 정상종료 될 때만 onComplete 호출
• onNext 대신 onSuccess 제공

Completable

• 데이터를 통지하지 않거나 에러 통지
• Completable 내에서 특정 부가 작용이 발생하는 처리 수행
• 해당 처리가 끝나면 onComplete, 에러시 onError 호출

다음은 Flow를 사용하는 sunflower의 PlantDao의 일부분이다.

@Query("SELECT * FROM plants ORDER BY name")
fun getPlants(): Flow<List<Plant>>

@Insert(onConflict = OnConflictStrategy.REPLACE)
suspend fun insertAll(plants: List<Plant>)

Room은 Flow뿐 아니라 Rx도 지원하므로, 위 코드는 다음과 같이 Rx를 사용한 코드로 변경할 수 있다.

@Query("SELECT * FROM Plant")
fun getPlants(): Flowable<List<Plant>>

@Insert(onConflict = OnConflictStrategy.REPLACE)
fun insertAll(plants: List<Plant>) : Completable

Rx의 연산자중 Completable은 데이터 자체를 반환하지는 않고 onComplete, onError만 알리기 때문에, 특별한 Response가 필요 없는 경우 사용하기에 적합하다.

db.plantDao().insertAll(plantList)
    .subscribeOn(Schedulers.io())
    .subscribe(
        { Log.d(TAG,"doWorkSuccess") },
        { Log.e(TAG, "Error seeding database - insert Error : ${it.message}") }
    )

스트림을 통해 받아온 데이터로 ui 작업을 해야 하는 경우,
다음과 같이 ui를 관리하는 스케줄러인 AndroidSchedulers.mainThread() 를 할당하여 사용한다.

plantRepository.getPlants()
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(
        { data -> _plantList.postValue(data) },
        { Log.e(TAG, "loadPlantList Error : ${it.message}") }
    )

한번의 Request & Response로 이루어져 일반적으로 한번의 값만을 방출하는 경우 (api 통신, client-server request/response 등) 다음과 같이 Flowable 대신 Single 혹은 Maybe 연산자를 사용할 수도 있다.

@Query("SELECT * FROM Plant WHERE plant_id = :plantId")
fun getPlant(plantId: String): Single<Plant>

Rx 스트림에 안드로이드 생명주기 연결하기

Flow는 asLiveData 혹은 repeatOnLifecycle 등을 이용하여 각종 컴포넌트의 생명주기에 쉽게 연결할 수 있지만, Rx는 직접 생명주기를 설정해주어야 한다.

Rx에서는 스트림 구독이 필요 없을 때, dispose할 수 있으며, 이를 위해 Disposable 이라는 인터페이스를 제공한다.
해당 객체를 이용하면 onComplete가 호출되지 않아도 dispose하여 강제로 중단시킬 수 있다.

fun main() {
    val observable: Observable<Long> = Observable.interval(1, TimeUnit.SECONDS)
    val disposable: Disposable = observable.subscribe { println(it) }

    Thread.sleep(3000)
    disposable.dispose()
}

실행 결과

0
1
2
Process finished with exit code 0

또한 observable이 여러개 존재할 경우 이를 한번에 관리하기 위하여 CompositeDisposable 를 사용할 수 있다.
여러 observable을 add로 추가하여 사용한다.

compositeDisposable.add(
    gardenPlantingRepository.createGardenPlanting(plantId)
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(
            { Log.d(TAG,"addPlantToGarden Success") },
            { Log.e(TAG,"addPlantToGarden Fail : ${it.message}")}
        )
)

안드로이드에서는 상황에따라 onDestroy 혹은 onPause에서 주로 dispose하여 생명주기와 연결시켜 관리할 수 있다.
Rx를 사용하는 모든 뷰에 연결할 것이므로 보일러플레이트를 줄이기위해 BaseView와 BaseViewModel을 만들어 작업하였다.

abstract class BaseFragment<VM : BaseViewModel> : Fragment() {

    abstract val viewModel : VM

    override fun onDestroy() {
        super.onDestroy()
        viewModel.clearCompositeDisposable()
    }
}

abstract class BaseViewModel : ViewModel() {
    protected val compositeDisposable = CompositeDisposable()

    fun clearCompositeDisposable() = compositeDisposable.clear()

    override fun onCleared() {
        super.onCleared()
        compositeDisposable.dispose()
    }
}