[연재] 코틀린 프로젝트 - 02 - 4단계: RxJava를 활용한 네트워크 서비스 비동기 태스크

이제 원격 서버에서 자료를 서로 주고 받기 위한 기법을 살펴보겠습니다. 메인 스레드인 UI 스레드와 서버에서 다운로드 같은 것을 하기 위한 일반 스레드를 서로 분리하고 비동기로 처리합니다. 처리한 결과를 UI에 반영하기 위해 서로 상태를 알 수 있어야 합니다. RxJava나 코틀린의 코루틴을 통해서 이러한 비동기 태스크를 만들어 낼 수 있습니다. 여기서는 https://www.themoviedb.org/ 사이트로부터 영화 정보를 가져오도록 서버를 연결하고 요청을 처리하는 부분을 만들 것입니다.

RxJava로 MovieManager 클래스 구현하기

먼저 RxJava를 통해 네트워크로부터 정보를 읽어올 서비스를 설계하고 이것은 일반 태스크로 구동하도록 합니다. 여기서 받아온 데이터는 앞서 만들어 놓은 데이터 클래스에 저장할 것입니다. RxJava 라이브러리를 통해서 이것을 구현하기 위해서는 UI 스레드는 뉴스를 가져올 서비스를 구독(subscribe()) 합니다. 이벤트가 발생하면 onNext()에서 처리할 것입니다. 

그림으로 나타내면 대략적으로 할 일들을 구분할 수 있습니다. 메인스레드는 Observer가 되며 일반 스레드의 서비스는 Observable이 되기 때문에 구독(subscribe())하게 되면 이벤트 발생에 따라 onNext()나 완료 시에는 onCompleted(), 에러 발생 시에는 onError()을 처리하게 됩니다. 

향후에는 서버에 접근에 RESTful API인 GET 방식으로 요청하면 서버는 응답에 대한 결과로 JSON을 돌려줄 것입니다. 이것을 데이터 객체로 변환 후 다시 RecyclerView의 아이템을 갱신해 UI를 다시 그리기를 하면 됩니다. 

1. 먼저 RxJava를 사용하기 위한 라이브러리를 build.gradle에 다음과 같이 추가합니다.

...
dependencies {
    //RxJava
    implementation "io.reactivex.rxjava2:rxjava:2.2.3"
    implementation 'io.reactivex.rxjava2:rxandroid:2.1.0'
...

2. 이제 패키지 하위의 ui/ 에 MovieManager.kt 파일을 새로 만듭니다. 작성 시 Observable은 [Alt+Enter]키를 누를 때 여러 개가 추천되는데 io.reactivex 패키지에 있는 것을 임포트합니다. 

...
class MovieManager() {
    fun getMovieList(): Observable<List<MovieItem>> { // Observable은 주로 생산자를 의미한다.
        return Observable.create { subscriber -> // 데이터 생성을 위한 create
            val movieList = mutableListOf<MovieItem>()
            for (i in 1..10) {
                movieList.add(
                    MovieItem(
                    1234,
                    5.0f,
                    "Test Title $i",
                    "2018-01-01",
                    "https://picsum.photos/480/640?image=$i",
                    "Test Overview"
                    )
                )
            }
            subscriber.onNext(movieList) // 구독자(관찰자)에게 데이터의 발행을 알린다.
            subscriber.onComplete() // 모든 데이터의 발행이 완료되었음을 알린다. 
        }
    }
}

서버에서 정보를 가져오기 위한 대략적인 구성은 먼저 Observable을 생성하고 반환하는 getMovieList() 메서드를 만듭니다. Observable.create() 메서드는 구독자 객체를 받을 수 있는 함수를 제공합니다. 구현은 람다식을 통해 구독자의 movieList데이터를 처리하도록 onNext()와 onComplete()를 호출할 수 있도록 합니다. onComplete() 이후에는 onNext()가 더 이상 사용될 수 없습니다.

• onNext(item: T): 뉴스가 새롭게 받는 이벤트가 발생하면 처리
• onComplete( ): 모든 데이터가 제공된 다음 호출 (onNext( ) 다음에 처리됨)
• onError(e: Throwable): API 호출 시 에러가 발생하면 이 메서드를 호출

3. 이제 MovieFragment.kt 파일에 다음과 같이 MovieManager 선언하고 뉴스를 가져올 때 초기화합니다. 

...
class MovieFragment : Fragment() {

    private val movieManager by lazy { MovieManager() } // (1) 늦은 초기화 선언
    // 혹은 lateinit var movieManager: MovieManager
...
    override fun onActivityCreated(savedInstanceState: Bundle?) {
        super.onActivityCreated(savedInstanceState)
...

        // 어댑터의 연결
        if (rv_movie_list.adapter == null) {
            rv_movie_list.adapter = MovieAdapter()
        }

        // 테스트용 데이터 생성 - (2) 기존 내용을 삭제하고 새로 작성
       requestMovie()
    }
    private fun requestMovie() {
        val subscription = movieManager.getMovieList()
            .subscribeOn(Schedulers.io()) // (3)
            .subscribe ( // 여기서는 데이터의 소비가 일어난다.
                { retrievedMovie -> // (4) onNext()의 처리
                    (rv_movie_list.adapter as MovieAdapter).addMovieList(retrievedMovie)
                },
                { e -> // (5) onError()의 처리
                    //Snackbar.make(rv_movie_list, e.message ?: "", Snackbar.LENGTH_LONG).show()
                    rv_movie_list.snackbar(e.message ?: "") // Anko의 스낵바
                }
            )
    }
}

(1)번과 같이 선언 시 lazy {...} 혹은 lateinit을 사용해 지연 초기화 할 수 있습니다. lazy와 lateinit은 차이점이 있습니다. 다시 정리해 보면 lateinit은 선언 시 다음과 같은 제약이 있습니다. 

•  값을 변경할 수 있는 var 프로퍼티만 사용 가능
•  non-null 프로퍼티만 사용 가능
•  커스텀 게터/세터가 없는 프로퍼티만 사용 가능
•  Int, Long등의 기본형 프로퍼티는 사용 불가능
•  클래스 생성자에서 사용 불가능
•  로컬 변수로 사용 불가능

따라서 lateinit은 Null이 될 수 없고 값이 변경될 수 있는 프로퍼티에 적용합니다. 초기화를 선언과 동시에 해줄 수 없거나 초기화를 미뤄야 할 때 사용하면 됩니다. 혹은 향후 Dagger라이브러리를 위해 @Inject 어노테이션을 사용하면 이 내용을 외부에서 주입 받을 때 lateinit을 통해 초기화를 지연시킬 수 있습니다. 

lazy {...} 기법도 지연시킬 때 사용하는데 람다식 매개변수를 받고 Lazy<T> 인스턴스를 반환 합니다. 이것의 제약 사항은 다음과 같습니다. 

• 값을 변경할 수 없는 val 프로퍼티에만 사용 가능
• 기본형에 사용 가능
• 커스텀 getter/setter가 없는 프로퍼티만 사용 가능
• non-null, nullable 둘다 사용 가능
• 클래스 생성자에서 사용 불가능
• 로컬 변수에서 사용 가능

몇 가지 lateinit과 반대되는 차이점이 있습니다. 특히, lazy는 블록 안에서 초기화 코드를 넣어 해당 코드가 사용되는 시점에 블록의 내용이 초기화 되게 됩니다. 또한 이 블록은 동기화를 제공하며 처리되기 때문에 멀티 스레드 환경에서 다중 접근 시 스레드에 안전한 블록이 됩니다. 

이제 (2)번에서 기존의 테스트용 데이터 생성 부분을 삭제하고 새롭게 작성된 부분을 분석해 봅시다.

requestMovie() 메서드에는 MovieManager를 사용하는 MovieFragment는 subscribe() 메서드를 호출하면서 이벤트 발생을 기다리는 '구독자' 즉 Observer가 됩니다. 

val subscription = movieManager.getMovieList().subscribe (
        { retrievedMovie -> 
           ... // (4) onNext 부분의 처리로 데이터를 소비하는 부분
        },
        { e ->  
            ... // (5) onError의 처리
        }

subscribe() 메서드는 (4)번과 (5)번 구현처럼 몇 가지 오버로딩된 메서드가 있는데 여기서는 다음 형식을 사용합니다. 

public final Subscription subscribe(
       final Action1<? super T> onNext, 
       final Action1<Throwable> onNext) {
       ...
}

따라서, 위와 같은 인자에 직접 람다식을 사용해 두 경우를 처리하도록 구현했습니다. onNext()에는 영화 정보를 받아온 어댑터를 설정하도록 하고, onError()에는 에러 발생 시 SnackBar와 같은 다이얼로그를 이용해 에러를 보여주도록 합니다. 이때 Anko 라이브러리를 사용해 축약할 수 있습니다.

이제 코드가 대략 구성되었습니다. 하지만 아직 UI 스레드에서 동작하기 때문에 새로운 스레드로 분리할 필요가 있습니다. 소스 코드 (3)번의 스케줄러에 대해 설명하겠습니다.

RxJava의 스케줄러

val subscription = movieManager.getMovie()
        .subscribeOn(Schedulers.io())
        .subscribe (...)

스레드로 분리하기 위해 movieManager의 구독 부분을 subscribeOn()메서드 체인을 하나 더 추가합니다. 이렇게 함으로서 람다식으로 구현되었던 Observable 코드는 분리된 스레드에서 처리되 수행됩니다. 인자로 넣어진 Schedulers는 분리된 스레드를 어떤 정책으로 실행할지 스케줄러에 주는 옵션입니다.

• io( ): 주로 IO 작업에 사용할 때
• computation( ): 연산 작업 위주의 태스크를 구성할 때
• newThread( ): 새로운 스레드를 생성해 처리
• test( ): 디버깅을 위해 

여기서는 Schedulers.io()가 사용되 IO 작업에 의도된 스레드가 만들어집니다. 사실 내부적으로 스케줄러는 정책을 세우고 구동하는데 IO는 입출력의 시점이 언제 시작하고 끝나는지 모르는 비 예측 태스크이기 때문에 스케줄러 입장에서는 우선순위를 낮추게 됩니다. 

반면에 연산 작업 위주의 computation()의 경우에는 일단 연산 된 시간 자체는 특별히 인터럽트가 없는 한 동일하기 때문에 스케줄러 입장에서는 실행시간을 예측할 수 있어 필요에 따라 우선순위를 높이거나 낮추는 정책이 사용됩니다. 새롭게 스레드를 생성하는 것은 기존의 스레드 풀을 사용하는 것 보다 오버헤드가 있을 수 있습니다. 

안드로이드를 위한 스케줄러 확장

이제 안드로이드에 특화된 스케줄러 정책을 위해 프로젝트에 build.gradle 의존성에 RxAndroid 라이브러리를 추가한 것을 기억할 것입니다. 

implementation 'io.reactivex.rxjava2:rxandroid:2.1.0'

4. 이 경우에는 다음과 같이 Observer의 스케줄러 옵션을 UI 스레드인 mainThread로 명시적으로 지정할 수 있습니다. 

val subscription = movieManager.getMovieList()
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread()) // rxandroid 에 의한 확장 - 새로운 스레드로 전환
        .subscribe ( ...
)

RxBaseFragment 클래스

이제 UI의 상태에 따라 앱이 가려지면 구독을 중단하고 다시 재개되면 구독을 다시 하도록 할 필요가 있습니다. 사용하지 않는 앱의 이벤트를 지속적으로 처리할 필요가 없기 때문 입니다. ui/에 새로운 파일 RxBaseFragment.kt 파일을 생성하고 다음과 같이 작성합니다.

...
open class RxBaseFragment : Fragment() {

    protected var subscriptions = CompositeDisposable()

    override fun onResume() {
        super.onResume()
        subscriptions = CompositeDisposable()
    }
   
    override fun onPause() {
        super.onPause()
        subscriptions.clear()
    }
}

5. 최종적으로 뉴스 요청을 위한 메서드로 감싸고 이 객체를 통해 Disposable 객체를 지정하도록 합니다. 

...
class MovieFragment : RxBaseFragment() {  // 기존의 Fragment()를 RxBaseFragment()로 교체
...
val subscription = movieManager.getMovieList()
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread()) 
        .subscribe ( ... 
)
subscriptions.add(subscription) // Disposable 객체의 지정

이제 실행을 해보면 테스트 데이터를 MovieManager를 통해서 일종의 관찰자인 MovieFragment에서 생산자이자 관찰 대상인 MovieManager의 객체를 생성하고 getMovieList()를 호출하면서 생산, 소비의 활동이 각 이벤트에 따라 처리할 수 있게 되었습니다. 

RxJava를 통한 비동기 태스크 구현에 대해 살펴봤는데 이 부분을 향후에 코루틴을 사용하는 방법으로 바꿀 수 있습니다.

이제 다음 단계에서 실제 데이터를 가져오도록 Retrofit을 사용해 HTTP 요청 처리 단계를 진행하고 적용해 봅시다.