안드로이드 프로젝트에 RxJava 적용하기

RxJava를 적용하여 비동기 작업이나 UI 이벤트를 효율적으로 처리해보자! 

 

1. RxJava란?

Reactive Extensions

RxJava를 알아보려면, 먼저 'Reactive Extensions'에 대해 알아야 한다. 얘는 ReactiveX라고도 부르며, 이벤트 기반 비동기 프로그래밍을 위한 라이브러리이다. Reactive Extensions은 매 이벤트마다 그에 대응하는 동작을 정의하는 기존의 이벤트 처리 방식인 콜백 방식과 달리 발생하는 이벤트를 이벤트 스트림에 전달하고, 이벤트 스트림을 관찰하다가 원하는 이벤트를 감지하면 이에 따른 동작을 수행하는 방식을 사용한다.

RxJava는 'Reactive Extensions' 라이브러리의 JVM 구현체로, Reactive Extensions 라이브러리에서 제공하는 기능들을 안드로이드를 포함 모든 자바 기반 플랫폼에서 사용할 수 있게 해준다.

 

RxJav가 특히 안드로이드 앱 개발자에게 사랑받는 이유는 다음과 같다.

1. 비동기 이벤트를 매우 쉽게 처리할 수 있다.
2. 이벤트나 데이터를 쉽게 가공 및 분배할 수 있다. 

 

Observable과 Observer

• 옵서버블(Observable): 이벤트를 만들어내는(emit) 주체로, 이벤트 스트림을 통해 만든 이벤트를 내보낸다.
• 옵서버(Observer): 옵서버블에서 만들어진 이벤트에 반응(react)하며, 이벤트를 받았을 때 수행할 작업을 정의한다. 이때, 옵서버가 옵서버블에서 만들어진 이벤트에 반응하려면 옵서버블에서 발생하는 이벤트를 옵서버가 관찰해야하는데, 이를  '옵서버가 옵서버블을 구독(subscribe)한다'라고 표현한다.

옵서버블과 옵서버의 관계

 

연산자

연산자(Operators)는 이벤트 스트림을 통해 전달되는 이벤트를 변환한다.

단순히 이벤트가 갖고 있는 값을 다른 형태로 바꿔주는 것뿐 아니라 특정 조건을 만족하는 이벤트만 이벤트 스트림에 흘려보내거나 이벤트의 개수를 바꿔주는 등 다양한 작업을 수행할 수 있다.

연산자는 중첩하여 사용할 수 있으며, 다음은 이벤트 스트림에 filter 연산자를 적용하여 2의 배수인 이벤트만 통과시키는 예 이다.

 

filter 연산자를 적용하여 2의 배수인 이벤트만 통과

 

스캐쥴러

스케쥴러(Scheduler)는 작업을 수행할 스레드(thread)를 지정한다. 안드로이드에서는 UI를 업데이트할 수 있는 스레이드인 메인 스레드를 사용할 수 있다.

스케쥴러는 observerOn() 메소드를 사용하여 지정하며, 이 메소드를 호출한 직후에 오는 연산자나 옵서버에서 수행되는 작업이 앞의 observerOn() 메소드에서 지정한 스레드에서 실행된다.

 

 

디스포저블

디스포저블(Disposable)은 옵서버가 옵서버블을 구독할 때 생성되는 객체로, 옵서버블에서 만드는 이벤트 스트림과 이에 필요한 리소스를 관리한다. 옵서버블로부터 더 이상 이벤트를 받지 않으려면 디스포저블을 통해 구독 해제(unsubscribe)를 할 수 있다. 옵서버블은 이를 감지해 자신을 구독하고 있는 옵서버가 더 이상 없는 경우 이벤트를 만들기 위해 유지하고 있던 리소스(ex: 뷰에 대한 참조 등)을 해제한다.

 

 

 

2. Retrofit에서 RxJava에 맞게 응답 전달하기

이번엔 Retrofit을 통해 받는 HTTP 응답을 옵서버블 형태로 받도록 수정하는 방법을 정리하겠다.

 

 

1) RxJava와 RxAndroid를 의존성에 추가

 

dependencies.gradle에 다음과 같은 코드를 추가한다.

ext { 
    ...
    // 프로젝트에서 사용할 RxAndroid, RxJava 버전 정의
    rxAndroidVersion = '2.0.1'
    rxJavaVersion = '2.1.3'
    ...
}

애플리케이션 빌드스크립트를 열어 다음과 같이 의존성을 추가한다.

...

android { 
    ...
}

dependencies { 
    ...
    
    // Retroifit에서 받은 응답을 옵서버블로 변환해주는 라이브러리를 추가
    implementation "com.squareup.retrofit2:adapter-rxjava2:$retrofitVersion"
    ...
    
    // RxAndroid와 RxJava 라이브러리를 추가
    implementation "io.reactivex.rxjava2:rxandroid:$rxAndroidVersion"
    implementation "io.reactivex.rxjava2:rxjava:$rxJavaVersion"
    
    ...
}

 

2) 데이터 처리부 수정하기

Retrofit에서 받은 응답을 옵서버블 형태로 반환하도록 하려면 두 가지 작업이 필요하다.

첫 번째, API가 구현된 인터페이스에서 각 API의 반환 형태를 Observable로 바꾼다.

interface NetworkService {

    @POST("users/signup")
    fun postSignup(
        @Header("Content-Type") content_type: String,
        @Body() body: JsonObject
    ): Observable<IntDataResponse> // 반환 타입 변경
  
}

두 번째, 앞에서 정의한 API를 호출할 수 있는 객체를 만들어주는 부분을 수정해야한다.

Retrofit에서 받은 응답을 옵서버블 형태로 변경해주는 RxJava2CallAdapterFactory를 API의 콜 어댑터로 추가하며, 비동기 방식으로 API를 호출하도록 RxJava2CallAdapterFactory.createAsync() 메소드로 콜 어댑터를 생성한다.

 private val retrofit: Retrofit =
        Retrofit.Builder().baseUrl(BASE_URL).client(okHttpClient)
            // 받은 응답을 옵서버블 형태로 변환해주도록
            .addCallAdapterFactory(RxJava2CallAdapterFactory.createAsync())
            .addConverterFactory(GsonConverterFactory.create())
            .build()

val service: NetworkService = retrofit.create(NetworkService::class.java)

 

3) 통신하는 부분 수정하기

앞에서 옵서버블을 사용하여 API 호출 결과를 받도록 수정했으므로, 기존에 이를 관리하기 위해 사용한 Call 객체를 디스포저블 객체로 바꿔야한다. 이때 생성되는 디스포저블 객체를 조금 더 효율적으로 관리하기 위해, 여러 디스포저블 객체를 한번에 관리할 수 있는 CompositeDisposable을 사용한다. 또한, 통신을 하는 부분에서는 옵서버블 형태로 반환되는 Response를 처리할 수 있도록 코드를 변경한다.

 

예를 들어 회원가입 통신을 진행 할 때, 회원가입이 제대로 완료되었는지를 SignupResponse의 status로 확인한다고 해보자.

기존 Activity에서 만들어줬던 signup() 메서드에서는 옵서버블 형태로 반환되는 SignupResponse를 처리할 수 있도록 코드를 변경해야한다. SignupResponse를 반환하는 옵서버블에 구독하면 디스포저블 객체가 생성되는데, 이때 생성된 디스포저블 객체는 CompositeDisposable에서 관리하도록 CompositedDisposable.add() 함수를 사용하여 추가하고, 받은 응답에서 SignupResponse만 뽑아 처리하도록 map() 함수를 사용하여 전달되는 데이터를 변경해야한다.

(책에서는 Acitivity에서 이 일을 모두 해결했지만, 내 프로젝트에서는 data 통신 관리를 위한 Repository 패턴과 MVVM 패턴을 적용했기 때문에 이에 맞게 코드를 작성하였다.)

 

 

data 부분

// SignupRemoteDataSource

fun signup(body:JsonObject) : Observable<IntDataResponse>

// SignupRemoteDataSourceImpl

override fun signup(body: JsonObject) =
        api.postSignup("application/json", body)
            .map { it }

// SignupRepository

class SignupRepository {
    val signupRemoteDataSource : SignupRemoteDataSource = SignupRemoteDataSourceImpl()

    fun signup(body : JsonObject) : Observable<IntDataResponse> = signupRemoteDataSource.signup(body)

}

 

 

ui 부분

SignupActivity에서 입력받은 ID와 password를 JsonObject 형태로 viewModel의 signup() 메소드로 전달한다.

class SignupViewModel : ViewModel(){

    private val signupRepository = SignupRepository()
    internal val disposables = CompositeDisposable()

    fun postSignupData(body: JsonObject){
        disposables.add(signupRepository.signup(body)
            .observeOn(AndroidSchedulers.mainThread())
            // 구독할 때 수행할 작업을 구현
            .doOnSubscribe {}
            // 스트림이 종료될 때 수행할 작업을 구현
            .doOnTerminate {}
            // 옵서버블을 구독
            .subscribe({
                // API를 통해 액세스 토큰을 정상적으로 받았을 때 처리할 작업을 구현
                // 작업 중 오류가 발생하면 이 블록은 호출되지 x

                // onResponse
            }){
                // 에러 블록
                // 네트워크 오류나 데이터 처리 오류 등
                // 작업이 정상적으로 완료되지 않았을 때 호출

                // onFailure
            })
    }

    override fun onCleared() {
        disposables.clear()
        super.onCleared()
    }
}

 

 

이렇게 비동기 이벤트를 쉽게 처리할 수 있고, 이벤트나 데이터를 쉽게 가공 및 분배할 수 있게 도와주는 RxJava에 대해 정리해봤다.

주로 RxJava는 통신을 하고, 통신 결과를 처리할 때 주로 많이 사용했는데 다른 곳에서도 쓸 수 있겠지? 많은 레퍼런스를 찾아봐야할 것 이다.