안드로이드 플랫폼은 태생부터 대규모의 이벤트 드리븐 애플리케이션을 일관적이고 신뢰도 높은 방법으로 구축하는 것이 힘든 구조입니다. 이번 GOTO Conference CPH 2015의 강연에서 RxAndroid의 창시자인 Matthias는 FRP(Functional Reactive Programming)을 선도적으로 도입한 기업인 SoundCloud의 안드로이드 애플리케이션 아키텍쳐가 어떻게 발전해 왔고 비동기적 이벤트 스트림에 RxJava를 사용하면서 어떤 변화가 있었는지 데이터 소스부터 비지니스 로직까지 아울러 설명합니다.

아래 비디오는 정확한 영어자막이 들어가 있으며 홈페이지에는 슬라이드가 동기화 되어있습니다. 기회가 되시면 자막을켜고 비디오를 직접 보는 것을 추천드립니다

소개 (0:00)

리액티브 프로그래밍, 특히 SoundCloud에서 안드로이드에 RxJava를 도입한 최근 몇 년간의 일을 말씀드리면서 강연을 시작하려 합니다. RxJava를 잘 모르신다면 인터넷에 자료가 많이 있습니다.

RxJava 라이브러리 자체보다는 RxJava가 안드로이드 프레임워크의 어떤 단점을 보완했는지, 그리고 우리 애플리케이션에 이를 어떻게 적용했고 개발에는 어떤 영향을 미쳤는지에 대해 초점을 맞추겠습니다. 이번 강연으로 문제 해결에 대한 인사이트를 얻을 수 있으면 좋겠습니다.

RxJava를 언제부터 사용했는지 살펴보다가 2013년 5월의 커밋을 발견했습니다. 배포전 버전부터 시작해서 이제 2년 반을 사용하고 있는 셈인데 그간 많은 변화가 있었습니다. 당연히 초반에는 몇몇 이슈가 발생했지만 나중에는 우리 애플리케이션에 막대한 영향을 미쳤기 때문에 결과적으로 RxJava 도입에 대해 모두 만족했을 겁니다.

이번 강연에서는 우리 애플리케이션 스택의 흥미로운 부분과 코드를 어떻게 분리했는지에 대해 말하고자 합니다.

애플리케이션 구성 (2:14)

애플리케이션은 어떻게 구성할 수 있을까요? 어떤 것들이 계층을 구성할까요? 저희도 2013년도에는 평범하게 애플리케이션 아키텍처에 대해 계층적으로 접근했습니다. 최상위에는 뷰나 프래그먼트와 같은 프리젠터가 있었습니다. 데이터 계층에서 표현 계층으로 데이터를 전달하는 비즈니스 객체는 중간층에 있었고요. 맨 아래에는 서비스 API나 로컬 저장소 등과 통신하는 데이터 객체를 놓았습니다.

2013년도 말에서야 이런 플랫폼 주도적 방식이 더이상 잘 작동하지 않음을 깨달았습니다. 안드로이드 팀, iOS 팀, 웹 팀이 각각 특정 기술 분야에 집중해서 플랫폼을 찾았죠. 사업적 가치에 의거한 특정 기능을 강조하는 상황에서 이런 시도는 잘 융화될 수 없었습니다.

그래서 우리가 전달하고자 하는 가치를 소프트웨어 자체가 반영할 수 있도록 조직을 기술 대신 기능 중심으로 분화했습니다. 백엔드 부분에서부터 시작했죠.

이런 변화를 적용하는 데는 시간이 필요했습니다. 모바일이 가장 나중이었기 때문에 몇 달 전까지만 해도 플랫폼 팀이 남아 있었지만, 앞으로 점차 매트릭스 조직으로 진화할 것입니다.

매트릭스 조직 안에서도 계층별 분화를 원할 수 있습니다. 프리젠터는 데이터에서 분리하고 싶겠죠. 동시에 애플리케이션 내에서 수직적으로 독립된 체계를 원할 수도 있습니다. SoundCloud의 기능을 예로 들어 보겠습니다. 트위터와 비슷한 타임라인 기능과 대비되는 새 콘텐츠 탐색이나 찾기 기능 등은 기능에 맞게 코드를 분리하고 싶을 겁니다.

계층 간 통신 방법 (5:04)

이런 수직적 방향에서는 계층간 통신이 필요합니다. 예를 들어 데이터 계층에서 비즈니스 로직을 거쳐 UI로 데이터를 옮겨야 합니다. 또한 다른 측면도 살펴봐야 합니다. 기능끼리 어떻게 통신할까요?

기능 매트릭스의 특정 원소들끼리 응집적이고, 시스템의 나머지 부분에서 독립적이길 원한다면 모든 원소들이 다른 모든 원소들에 접근 가능한 방식으로 구현해서는 안됩니다.

RxJava를 활용하면 이런 문제를 쉽게 해결할 수 있습니다. 안드로이드에서 애플리케이션의 컴퍼넌트끼리 통신하는 메커니즘은 여러 개가 있는데요. 흔히 쓰이는 인텐트나 핸들러 메시지를 예로 들어보죠. 객체 간의 콜백은 구현 방식이 좀 복잡하고 잘 구현하기 어렵습니다. 특히 JSON 응답을 가져오는 경우와 같이 기능이나 계층 간의 트랜지션이 일어나는 접점에서는 동시 실행이 자주 일어납니다. 자연스레 이런 문제를 재차 해결하는 것이 좋은 접근법인지 의문이 들게 되죠.

사운드 스트림 (6:38)

RxJava를 사용하여 우리 앱의 사운드 스트림 기능을 구현하면서 이런 문제를 어떻게 해결했는지 말씀드리겠습니다. 사운드 스트림은 처음 앱을 열면 보이는 화면이자 주로 제가 일반적으로 구현하게 되는 화면입니다. 나중에는 그저 일감 목록처럼 느껴지는데 카드뷰이든 그리드뷰이든 비슷하게 같은 문제의 연속인 셈이죠.

여러 개를 나열한 목록인 사운드 스트림은 평범해 보입니다. 그러나 아이템들이 무한으로 스트리밍되기 때문에 사실은 사실은 꽤 복잡합니다. 매 분 혹은 매 시간마다 새로운 콘텐츠가 발생해서 점차 방대해지므로 페이징처럼 데이터 전송을 늦추거나 딜레이를 주는 방법을 고민해야 합니다. 맨 밑까지 스크롤하면 그 때 새 컨텐츠를 받아오는 식이죠. 사용자들은 당겨서 새로고침같은 방식들이 친숙합니다. 한편 다음 페이지 로딩에 실패하는 등의 에러 상태도 고민해야 합니다. 재시도 로직을 넣을 수도 있겠죠.

이런 문제를 한꺼번에 해결하기 위해 지저분한 코드를 작성하는 개발자가 많을 겁니다. RxJava는 이런 문제를 깔끔하게 해결하는데 큰 도움이 됩니다. ✨

사운드 스트리밍 분리하기 (8:46)

이 페이지는 표현 레벨, 비즈니스 로직 레벨, 데이터 레벨에서 볼 수 있는 여러 특정 타입의 객체로 구성됩니다.

제일 윗 부분에는 안드로이드 뷰와 프래그먼트 같은 프리젠터가 있죠. 우리 애플리케이션에서는 MVP를 사용했으므로 뷰가 매우 가볍습니다. 스크린을 표현하는 객체가 화면을 구성하고 비즈니스 로직과 통신하며 안드로이드 LifeCycle 이벤트를 listen합니다.

기능 오퍼레이션, 유즈 케이스 클래스 혹은 interactor라고 불리는 객체는 비즈니스 로직을 포함합니다. 다중의 데이터 소스를 엮고 스케줄링도 하죠. 무거운 작업은 메인 UI 스레드에서 하지 말아야 합니다. 표현 계층 보다는 기능 오퍼레이션 계층에서 이런 작업과 비즈니즈 규칙 등을 담당하게 합니다.

저희는 RxJava를 통해 연결 고리를 만들었습니다. RxAndroid 라이브러리를 조금 사용했고, 안드로이드 메인 스레드에 콜백을 보낼 수 있는 핸들러 스레드 스케줄러를 주로 활용해서 이를 서버나 일반적인 데이터 소스로부터 온 데이터를 담는 다운스트림으로 사용했습니다.

제일 아래 계층은 데이터 계층으로 특정 기능과 관련된 데이터를 모두 담고 있습니다. 예를 들어 사운드 스트림에서는 정기적으로 동기화한 컨텐츠를 넣은 로컬 데이터베이스와, 이 데이터를 가져온 서버 API가 해당됩니다.

비즈니스 로직이 UI와 대면하는 접점은 하나여야 한다는 점을 강조하고 싶어요. UI가 여러 소스로부터 데이터를 가져오기 위해 다양한 객체를 조작한다면 뷰는 정말 복잡해지고 테스트하기 어려울 겁니다. 뷰가 단일 접점만 가지는 방식을 택한다면 테스트하기 쉬운 아래쪽 스택에서 로직을 구현할 수 있습니다.

뷰와 프리젠터 (12:07)

class SoundStreamFragment extends LightCycleSupportFragment {

  @Inject @LightCycle SoundStreamPresenter presenter;
  
  public SoundStreamFragment() {
    setRetainInstance(true);
    ... 
  }
  ...
}

뷰의 예시입니다. 프래그먼트를 뷰처럼 취급하기 때문에 컨트롤러 측면의 기능은 없습니다. 액티비티나 프래그먼트를 컨트롤러로 사용하는 MVC를 선호하는 분도 계시겠지만 우리에게는 프레임워크 간 접착제이자 앱을 구성하는 코드의 일부분일 뿐입니다. 저희는 프래그먼트를 많은 부분 분리하고 적은 것만 포함하며 단순한 묶음 용도로 사용합니다.

뷰를 조종하고 비즈니스 로직과 통신하는 프리젠터를 주입하기 위해 직접 구현한 LightCycle이라는 라이브러리를 사용하는데 사실 RxJava로 구현하지는 않았습니다. LightCycle은 안드로이드에서 받은 LifeCycle 이벤트를 프래그먼트 내의 여러 협력자에게 보내는 역할을 합니다. 이는 하나 혹은 다수의 프리젠터가 될 수도 있고, 심지어 여러 LightCycle을 엮어서 사용할 수도 있으며 뷰를 트래킹할 수도 있습니다. 프래그먼트는 주입 상태를 유지하기 위해 계속 유지되는데 이를 통해 화면 방향이 전환되는 등의 상황에서도 저렴한 비용으로 백그라운드에서 가져오는 데이터 손실을 방지할 수 있습니다.

LightCycle은 프래그먼트 내의 dispatcher와 협력해서 동작하며, 프래그먼트 자신이 dispatcher가 될 수도 있습니다. 프래그먼트가 시스템으로부터 onCreate 같은 콜백을 받으면 이 이벤트를 listen 하고 주입한 다른 LightCycle에게 전달합니다. 만약 한 프래그먼트에 세 개의 LightCycle를 주입했다면 LightCycle끼리는 서로 알 필요가 없습니다. 단일 클래스의 응집성을 높이지만 프래그먼트로서 같은 API를 노출하게 됩니다. 액티비티에서도 마찬가지로 작용합니다.

격리 상태에서 테스트할 수 있어서 독립적인 유닛 테스트가 가능합니다. 프레임워크와 클래스 사이의 중개자로 LightCycle을 사용하므로 유닛 테스트를 개별 프래그먼트 단위가 아니라 LightCycle과 밑단 로직 단위로 실행합니다.

class SoundStreamPresenter extends RecyclerViewPresenter<StreamItem> {
  ...
  @Override
  protected CollectionBinding<StreamItem> onBuildBinding(Bundle args) {
    return CollectionBinding.from(
      streamOperations.initialStreamItems())
        .withAdapter(adapter)
        .withPager(streamOperations.pagingFunction())
        .build();
  }
}

사운드 스트림을 표현하는 화면 프리젠터의 예제입니다. 200줄쯤 되는 코드이므로 작지도 무겁지도 않습니다. 어떤 목록을 그릴 경우 바인딩 스텝을 거치는데 이는 MVVM이나 안드로이드의 데이터 바인딩 라이브러리에서의 데이터 바인딩과는 좀 다릅니다. 바인딩은 RxJava로 구축한 Observable 시퀀스를 UI에 연결하는 단계로 많은 화면에서 재사용가능한 기반 코드입니다.

CollectionBinding에도 주목할만 합니다. ‘뷰가 생성됐으니 이 뷰를 위한 데이터를 가져와 달라’는 콜백을 LightCycle이 프리젠터에게 보내려면, 이 프리젠터를 위한 유즈 케이스 클래스인 스트림 오퍼레이션에 말하는 방식으로 구현합니다. x를 y 어댑터에 z 빌더 구문을 통해 올리라고 말하는데 페이징 기능도 이 방식으로 구현할 수 있습니다.

다른 방향도 잘 작동합니다. 프리젠터는 클릭 이벤트 등을 listen 할 수 있어서 뷰에 버튼이 있다면 자신이나 내부의 subscriber를 클릭 이벤트에 붙여서 비즈니스 로직에 전달할 수 있습니다.

페이징 (16:59)

페이징 문제는 많은 개발자들이 해결해야할 문제로 저도 예전에 구현할 때 어려움을 겪었습니다. 계층 사이에서 벌어지는 문제이므로 뷰 단의 로직과도 관련이 있지만, 특정 페이지를 준 후 다음 페이지로는 어떻게 가야 할 지와 같이 결정과 관련된 로직도 있습니다. 예를 들면 HTTP 링크를 통해 네트워크에서 받아온 JSON 객체의 다음 페이지로 가는 문제이죠.

이런 일, 즉 다음 페이지의 아이템을 위해 쿼리를 생성하는 법이나 데이터 커서를 받아오거나 옮기는 것을 뷰에서 처리해서는 안됩니다. 저희 회사는 데이터를 받아오는 방법이 뷰에 노출되도록 정보를 만드는 것이나 재사용 가능한 것을 만드는 것을 엄격하게 관리하는데, 그 이유는 페이징을 사운드 스트림만 하는 것이 아니기 때문입니다.

Rx를 사용하면서 우리는 Rx가 퍼블리싱하는 Subject와 유사한 추상화 클래스인 pager를 만들었습니다. 한 쪽에는 이벤트를 붙이고 다른 한 쪽에는 다른 리스너가 구독하도록 해서 단방향 통신을 하는 이벤트 채널처럼 동작합니다. 프래그먼트는 페이저를 구독하고 유효한 페이지가 이미 있다면 subscriber에게 Rx onNext 호출을 통해 그 페이지를 전달할 수 있습니다.

두 번째 로직은 페이지의 맨 아래까지 스크롤하면 다음 페이지를 로딩할 시점임을 감지하는 것입니다. 재사용 가능한 스크롤 리스너는 페이저에게 next라는 신호를 보내서 다음 페이지로 이동하게 합니다. 스크롤 리스너는 다음 페이지를 로드하는 방법을 Rx의 switchOnNext라는 오퍼레이터를 통해 알 수 있습니다. 이를 통해 아이템 스트리밍을 주고 다중 스트림 사이에서 전환할 수 있습니다.

switchOnNext를 사용해서 첫 번째 아이템 스트림을 주면 다음 페이지로 전환할 수 있지만, 시퀀스를 실제로 listen하는 호출자 입장에서는 단일한 아이템 스트림처럼 여길 수 있습니다. 내부적으로는 목록 사이에서 포인터를 움직이는 원리와 같습니다. 페이저는 현재 아이템 스트림이 무엇인지와 다음 스트림이 무엇인지에 대한 참조를 가집니다. switchOnNext를 사용하면 상태 변화를 조작하지 않고도 쉽게 이를 구현할 수 있죠. RxJava가 여러 subscribe에게 아이템 페이지를 발행하는 어려운 부분을 대신 해줍니다.

다음 페이지를 어떻게 얻을지는 페이징 함수에 요청해서 결정합니다. 페이징 함수를 요청하는 프리젠터 샘플 코드를 보면 특정 아이템 페이지와 다음 페이지를 어떻게 갈 수 있는지 알려주는 페이징 함수를 비즈니스 로직에게 요청하는 것을 볼 수 있습니다. 사운드 스트림의 경우 백그라운드에서 데이터를 동기화하기 때문에 데이터베이스 호출의 형태가 되며, 재사용 가능합니다. 단지 페이징 함수만 페이저에게 제공하면 되므로 이런 객체를 다른 화면에 구현해서 동작시킬 수 있습니다.

유즈 케이스 (21:03)

class SoundStreamOperations {

  Observable<List<StreamItem>> initialStreamItems() {
    return loadFirstPageOfStream()
      .zipWith(
        facebookInvites.loadWithPictures(),
        prependFacebookInvites())
          .subscribeOn(scheduler);
  }

  ...
}

아랫단으로 더 내려가 보면 UI로 데이터를 전달하는 유즈 케이스 클래스들이 있습니다. 앞서 스트림화된 아이템 목록인 페이지가 필요했었죠. 이런 여러 가지가 화합되는 계층이 바로 여기입니다.

첫 번째로 수직적 커뮤니케이션이 있는데, Observable 시퀀스를 다시 반환해서 UI가 전송된 데이터를 구독하고 받을 수 있도록 합니다. 한편 수평적 커뮤니케이션도 있는데 zipWith 오퍼레이션을 사용해서 관계 없는 기능에 도달해서 facebookInvites 요소인 이 화면을 구성하기 때문입니다. 애플리케이션의 신규 사용자라면 친구를 초대하는 액션을 호출하게 됩니다. 어디서 데이터가 오는지나 어떤 조건에서 보이게 되는지와 같은 모든 관련 사항에 대해 이 특정 클래스가 알 필요는 없습니다. 깔끔한 Rx 아키텍처처럼 사운드 스트림은 독립된 기능으로 작동합니다. 단일 스트림이 있을 뿐이며 모든 복잡한 로직은 하단에 감춰집니다. 물론 스케줄링도 시퀀스에게 언제 실행되는지 알려주는 subscribeOn를 호출해서 스케줄러를 주입하면 그만입니다.

기능 데이터 (23:05)

class SoundStreamStorage {

  Observable<PropertySet> streamItems(int limit) {
    Query query = Query.from("SoundStreamTable").limit(limit);
    return propellerRx.query(query).map(new StreamItemMapper());
  }

  ...
}

아직 데이터를 받아오는 과정이 남았습니다. 이 유즈 케이스 클래스는 여러 협력 클래스에게 데이터를 요청해야 합니다. 사용자가 가능한 빠른 시간에 볼 수 있도록 데이터 동기화를 합니다. 사운드 스트림의 경우 아이템 페이지를 불러오려면 데이터베이스로 접근해야 하겠죠. 이 때 데이터베이스에 직접 접근하지 않도록 추상화하여 저장 클래스처럼 사용하는 협력 클래스가 있습니다. 호출하는 입장에서는 이 협력 클래스가 어디에서 데이터를 가져오는지 알 수 없고 커서 객체도 보이지 않으므로 데이터베이스가 상위 계층으로 노출되는 것을 막을 수 있습니다.

우리 회사에서는 안드로이드 SQLite 데이터베이스를 추상화한 Propeller라는 라이브러리를 사용합니다. 테이블을 컬렉션과 이터러블로 접근하므로 리액티브 프로그래밍에 적합합니다. Guava나 Java의 표준 컬렉션 메커니즘을 사용해서 로컬 저장소의 데이터를 다룰 수 있도록 이터러블 커서를 제공하며, Observable과 이터레이터 간의 이중 원칙이 있어서 이들을 Rx Observable 시퀀스로 변환하는 것이 매우 쉽습니다.

작성한 쿼리를 받아서 Observable 시퀀스로 변환해주는 Rx 컴패니언 객체를 Propeller에서 사용합니다. 이 때 만들어지는 커서나 테이블의 열을 특정 아이템에 매핑할 수 있습니다.

전체 계층에서 일어나는 수직적인 작업은 이런 모습입니다. 이제 서로 관련이 없는 기능간에도 커뮤니케이션할 수 있는 수평적인 작업에 대해 말씀드리겠습니다.

크로스 스크린 메시징 (25:28)

크로스 스크린 메시징은 정말 자주 사용되는 케이스로 대부분 다뤄본 경험이 있을 겁니다. 사운드 스트림의 경우 다른 사람들이 포스트하거나 리포스트한 트랙과 재생 목록이 섞여 있는데다 서로 상호작용합니다. 이 곳 화면 중간에서 재생 목록을 클릭하다면 새로운 기능이 보여지죠.

좋아요 버튼을 누르면 카운트가 증가하고, 뒤로가기 버튼을 누르면 이전 화면으로 돌아갑니다. 카운트의 작은 하트모양 아이콘을 누르는 즉시 카운트가 증가합니다. 각기 다른 화면의 객체가 반응하는 셈이죠.

사운드 스트림은 다른 팀에서 만든 앱이므로 다른 기능이 너무 긴밀하게 작용해서는 안됩니다. 다른 팀에서 만든 두 화면을 나란히 놓고 생각해 봅시다. 두 화면 사이의 커뮤니케이션 채널을 어떻게 만들 수 있을까요?

두 화면이 서로 알게 하지 않고도 쉽고 효율적인 방법으로 데이터를 전달할 수 있습니다. 안드로이드는 인텐트와 같은 방법을 제공하지만 parcelable처럼 복잡한 기능을 구현해야 하므로 이를 좋은 기능이라고 생각하지 않습니다. 인텐트로 데이터를 전달하려면 parcelable API의 제약을 받는 bundle에 데이터를 넣어야 하므로 프로세스간 통신을 하지 않는 경우에도 marshalling을 할 수 있어야 합니다. 프로세스간 통신을 하는 경우가 더 적은 것을 생각하면 비효율적이죠. 한 화면에서 다른 화면으로 전환한 것 뿐인데 이런 작업을 해야 하다니 어색합니다.

화면에서 화면을 업데이트하는 예제 (27:56)

우리는 이 문제를 Rx Subject를 사용해서 해결했습니다. 이제 액티비티를 시작할 때 어쩔 수 없이 보내는 인텐트를 빼고는 데이터 전송에는 인텐트를 사용하지 않습니다. Subject는 EventBus와 같은 단일 이벤트 큐라고 이해하면 됩니다.

Observable<PropertySet> toggleLike(Urn urn,
                                   boolean addLike) {
 return storeLikeCommand.toObservable(urn, addLike)
          .map(toChangeSet(targetUrn, addLike))
          .doOnNext(publishChangeSet);
}

우리 코드의 예제를 볼까요? 좋아요를 트리거하는 막대를 포함한 이 특정 화면을 위한 유즈 케이스 클래스의 작동 과정입니다. 좋아요를 타겟하기 위해 데이터베이스에 스택을 유지해야 하므로 데이터베이스에서 일정한 change set을 유지할 수 있는 명령 클래스가 필요합니다. 이 클래스를 Rx Observable 시퀀스로 바꾸는데 성공하면 map 오퍼레이터를 사용해서 애플리케이션에서 사용할 수 있는 객체로 변환시킵니다.

이런 사실에 관심있는 다른 화면에 알려주기 위해 맵 오퍼레이터를 사용해서 이를 change set으로 바꾼 후 이 change set을 애플리케이션에 발행했습니다.

publishChangeSet: Action1(PropertySet)

@Override
public void call(PropertySet changeSet) {
  eventBus.publish(
    EventQueue.ENTITY_STATE_CHANGED, /* <-- RxSubject in disguise! */
    EntityStateChangedEvent.fromLike(changeSet)
  );
}

발행은 데이터를 건네받는 Rx 액션으로 eventBus를 사용해서 배포했습니다. eventBus는 Rx Subject로 애플리케이션의 다양한 곳에서 싱글톤으로 사용할 수 있습니다. 유닛 테스트도 가능해서 eventBus에 가짜 이벤트를 보내서 다른 컴퍼넌트가 제대로 이벤트를 받고 구독과 구독 취소를 할 수 있는지 확인할 수 있습니다.

SoundStreamPresenter

protected void onViewCreated(...) {
  eventBus.subscribe(
    EventQueue.ENTITY_STATE_CHANGED,
    new UpdateListSubscriber(adapter)
  );
}

받는 쪽은 뷰 계층에 위치합니다. 이벤트를 잘 받았나 확인하기 위해 이벤트를 받으면 뷰가 업데이트돼야 하죠. 같은 이벤트 큐를 구독하고 어댑터를 업데이트 했습니다. change set을 받으면 리스트의 특정 아이템을 업데이트해서 그리는 것처럼 쉽게 업데이트할 수 있습니다.

구현 패턴 (31:41)

생명 주기에 따른 구독

private CompositeSubscription viewLifeCycle;

protected void onViewCreated(...) {
  viewLifeCycle = new CompositeSubscription();
  viewLifeCycle.add(...);
  ...
}

protected void onDestroyView() {
  viewLifeCycle.unsubscribe();
}

Observable 시퀀스를 어떻게 프레그먼트나 액티비티와 같은 뷰에 연결할 수 있을까요? 가장 간단한 방법은 LifeCycle 콜백에서 구독하는 것입니다. viewLifeCycle를 따라 onViewCreated 호출시에 시퀀스를 구독하고 onDestroyView 콜백을 받을 때 구독을 취소하면 됩니다. 독립된 Rx 이벤트의 콜백을 이용하는 방법도 있지만 좀 복잡하므로 쉬운 방법을 택했습니다.

빠른 패스와 느린 업데이트

Observable<Model> maybeCached() {
  return Observable.concat(cachedModel(), remoteModel()).first()
}

빠른 패스나 느린 업데이트는 ‘concat 부터 하는’ 접근 방법입니다. 많은 개발자들이 캐싱된 값을 UI에 가능한 빠르게 보내서 UI의 반응성을 높이도록 구상하지만 결국은 캐싱하지 않고 네트워크에서 받아오는 길을 택하는 경우가 많습니다. 이제 RX의 연쇄적인 오퍼레이터를 사용해서 concat 부터 하는 접근 방법을 사용할 수 있습니다. 각각에 두 시퀀스를 덧붙여서 첫 번째가 먼저 완료되거나 아무 값도 받지 않으면 다음 시퀀스로 넘어가게 할 수 있습니다. 혹은 맨 뒤에 first를 더해서 캐시에서 값을 받으면 이 아이템을 사용하고 인터넷 연결은 하지 않도록 할 수도 있습니다. 데이터를 캐시 이벤트에서 받아오는 기능을 구현할 때 자주 사용하는 패턴입니다.

느린 업데이트처럼 ‘발행을 먼저’ 할 수도 있습니다. 즉 캐시로부터 먼저 발행해서 뷰가 즉시 아이템을 받을 수 있도록 하는 것입니다. 그러나 그 후 업데이트된 정보를 가져와서 다시 발행해야 하므로 몇 초 후 숫자가 바뀌는 결과를 볼 수도 있습니다. getOrElse의 리액티브 버전인 셈입니다.

Observable transformer

Observable<Model> scheduledModel() {
  return Observable.create(...).compose(schedulingStrategy)
}

class HighPrioUiTask<T> extends Transformer<T, T> {
  public Observable<T> call(Observable<T> source) {
    return source
      .subscribeOn(Schedulers.HIGH_PRIO)
      .observeOn(AndroidSchedulers.mainThread())
  }
}

Observable transformer도 정말 훌륭합니다. compose 오퍼레이터를 사용하고 한 시퀀스 내의 구성 단계에서 여러 숫자를 받아서 단일하게 변환시킵니다.

예를 들어 백그라운드 스레드에서 매번 subscribeOn를 부르고 안드로이드 메인 스레드에서 observeOn를 부르는 대신 transformer를 선언해서 해당 기능을 수행하도록 할 수 있습니다. 그 다음 우선 순위가 높은 UI 태스크처럼 이 transformer의 객체를 Observable 내에 구성할 수 있습니다. 이 스케쥴링 모델 시퀀스를 유닛 테스트할 경우 이를 유닛 테스트 내에서 스왑하고 현재 스레드와 동시에 실행할 수 있습니다.

실행 지연

Observable<Integer> intSequence() {
  return Observable.create((subscriber) -> {
    List<Integer> ints = computeListOfInts(); /* <-- expensive! */
    for (int n : ints) {
      subscriber.onNext(n);
      subscriber.onCompleted();
    }
  }
}

Observable<Integer> intSequence() {
  return Observable.defer(() -> {
    return Observable.from(computeListOfInts());
  }
}

데이터 계층의 Propeller에서 실행을 지연시키는 경우도 있습니다. integer 목록을 계산하는 고비용의 함수가 있다고 가정한다면 이를 백그라운드 스레드에서 실행하겠죠. 실제로 이 데이터를 구독하기 전까지는 이 함수의 호출을 지연하는 것이 좋고, 이 함수는 리스트를 다루므로 반복적인 동작을 할 겁니다. 보통 Observable을 생성하고 계산 결과를 받아 subscriber에게 발행하게 됩니다. 그러나 이 코드는 에러와 backpressure를 잘 핸들링하지 않는다는 문제점이 있습니다.

대신 이런 어려운 문제를 대신 해주는 RxJava 오퍼레이터를 재사용하는 것을 권장합니다. defer라는 오퍼레이터는 Observable 팩토리인 lambda를 전달할 수 있습니다. subscriber가 생기면 계산 결과를 직접 from이라는 팩토리 메서드에 전달하면 되고, from은 어떤 자바 컬렉션이든지 Observable 시퀀스로 변환합니다. 단일 통로를 사용할 수 있으며 RxJava가 제공하는 모든 기본 기능을 재사용할 수 있어 효율적입니다.

흔히 범하는 실수 (36:40)

전달 인자를 빠뜨리고 구독하기

Observable.create(...).subscribe(/* no-args */)

–> OnErrorNotImplementedException

먼저 전달 인자가 없는 subscribe 호출에 시퀀스를 지나치게 의지하는 경우가 있습니다. 문제를 인지하지 못할 수 있으므로 저는 이를 사용하지 않습니다. 에러를 받지 않는다는 가정 하에 작업하다가 치명적인 예외가 발생해서 애플리케이션에 크래시가 나는 경우가 생길 수 있습니다. 시도만 해보고 결과에는 관심이 없는 호출을 하는 경우라도 항상 시퀀스를 구독하고 에러 처리를 하세요. 방심하다가는 출시된 제품에서 크래시가 발생할 수 있습니다.

ObserveOn: onError

Observable.create((subscriber) -> {
   subscriber.onNext(value); /* <-- gets dropped! */
   subscriber.onError(new Exception());
}.observeOn(mainThread()).subscribe(...)

–> onError가 먼저 발생해 onNext를 중단시킵니다.

subscriber나 Observable이 있는 상황에서 즉각 값을 발행했는데 무엇인지 모르지만 실패했다고 가정해 봅시다. 예외와 함께 onError를 호출해서 메인 스레드의 subscriber가 값을 먼저 받은 후 에러를 받기를 기대하겠지만 실제로는 onError가 먼저 발생해서 onNext가 무시되기 때문에 그렇게 되지 않습니다. observeOn이 시퀀스가 실패했음을 알아차리는 순간 subscriber에게 뭔가 잘못됐음을 먼저 알려서 구독 해제를 하도록 하기 때문에 다른 아이템은 발행하지 않습니다. 따라서 예측하지 못했던 결과가 발생할 수도 있으니 조심해야 합니다.

ObserveOn: Backpressure (38:30)

public void onStart() {
  request(RxRingBuffer.SIZE); /* 16! */
}

public void onNext(final T t) {
  ...
  if (!queue.offer(on.next(t))) {
    onError(new MissingBackpressureException());
    return;
  }
  schedule();
}

Backpressure는 방대한 주제이므로 자세히 다루지는 않겠지만 문제를 일으킬 수 있다는 점을 주의해야 합니다.

observeOn 오퍼레이터의 코드를 보여드릴텐데, 기본적으로 RxJava가 내부적으로 push 모델에 기반하지 않기 때문에 request 호출 부분을 강조하고 싶습니다. 이는 Observable은 빠르게 알림을 발행하는데 subscriber가 너무 느리게 알림을 처리하는 경우를 해결하기 위해 고안된 “reactive pull”을 기반으로 하고 있습니다. 내부적으로 request는 일정량의 아이템을 요청하는데 만약 옵저버가 이들을 처리하지 않은 상태에서 다음 아이템 묶음을 요청한다면 큐에 생성된 새 아이템을 가져오는데 실패하게 됩니다. 특히 버전 16의 경우 안드로이드의 내부 버퍼 사이즈에서 문제가 발생하는데 16개의 아이템을 빠르게 소모해버리면 missing backpressure exception이 발생해서 애플리케이션에 크래시가 발생합니다.

이런 문제를 해결하려면 대상 스레드의 로드를 줄여서 부하가 걸리지 않도록 해야 합니다. 특히 모든 draw 호출이 집중되는 안드로이드의 메인 루퍼를 조심해야 하죠. 저희 회사에서는 방어적인 전략을 써서 메인 스레드에서 일어나는 모든 작업을 스케줄링합니다. 메인 스레드에 observeOn를 쓰면 더 이상 동시성에 대한 고민없이 간단하게 구현할 수 있습니다. 물론 이에 따른 비용이 발생하는데, 메인 스레드와 메인 루퍼에 너무 부하를 걸면 안드로이드가 메시지를 처리하는 것보다 다른 곳에서 전달되는 작업 속도가 빨라지는 시점에 문제가 발생하게 됩니다.

toList 등을 사용해서 시퀀스를 버퍼링하거나 buffer 오퍼레이터를 사용해서 시퀀스를 받아서 더 큰 묶음으로 발행할 수도 있습니다. 내부적인 backpressure 오퍼레이터와, 오버라이딩해서 내부 버퍼 사이즈를 늘릴 수 있는 시스템 프로퍼티도 있습니다.

디버깅 (41:15)

뭔가 잘못된다면 어떻게 해야 할까요? 다행히도 디버깅할 이슈는 대부분 subscriber에게 메시지가 전달되지 않는 경우입니다. RxJava의 버그를 발견하는데 꽤 시일이 걸렸죠.

Observable.just(1, 2, 3)
    .map((n) -> {return Integer.toString(n);}
    .observeOn(AndroidSchedulers.mainThread());

여러 개의 숫자를 발행하고 스트링으로 변환하는 예제 시퀀스입니다. 매핑 단계 이후에 observeOn이 있는 이 예제를 어떻게 디버깅할까요?

물론 디버거를 사용할 수도 있지만, 내부적인 backpressure 관련 요소와 함께 observeOn을 보는 것이 난해하므로 좀 더 쉬운 방법이 필요합니다.

따라서 우리는 어노테이션으로 바이트 코드를 주입하는 작은 라이브러리, Gandalf를 만들었습니다. 처음에는 애플리케이션에 로그를 주입하는 Jake Wharton의 “Hugo” 프로젝트로부터 시작했는데, 여기에 어노테이션 범위를 늘려서 적용했습니다. 로그뿐만 아니라 애플리케이션에 다른 수단을 바이트 코드로 주입할 수 있도록 했습니다.

로그 출력을 위해서는 @RxLogObservable과 @RxLogSubscriber 어노테이션을 사용하며 예제는 다음과 같습니다.

@RxLogObservable
Observable<String> createObservable() {
  return Observable.just(1, 2, 3)
    .map((n) -> {return Integer.toString(n);}
    .observeOn(mainThread());
}

@RxLogSubscriber
class StringSubscriber extends Subscriber<String> {}

Gandalf 플러그인을 스크립트에 적용하고 @RxLogObservable 어노테이션만 붙이면 됩니다. subscriber에도 적용할 수 있는데 결과는 다음과 같습니다.

[@Observable :: @InClass -> MainActivity :: @Method 
    -> createObservable()]
[@Observable#createObservable -> onSubscribe() ::  
    @SubscribeOn -> main]
[@Observable#createObservable -> onNext() -> 1]
[@Observable#createObservable -> onNext() -> 2]
[@Observable#createObservable -> onNext() -> 3]
[@Observable#createObservable -> onCompleted()]
[@Observable#createObservable -> onTerminate() :: 
    @Emitted -> 3 elements :: @Time -> 4 ms]
[@Observable#createObservable -> onUnsubscribe()]

단순히 커맨드 라인에서 받은 시퀀스를 출력하는 예제입니다. MainActivity 클래스의 createObservable 메서드에서 메인 스레드를 구독합니다. 1, 2, 3 등의 값을 받다가 완료하고 종료하면 전체 시간을 알려줍니다. 마지막에는 unsubscribe를 호출합니다.

몇 번째 단계에서 구독이 해제되는지 확인할 수 있으므로, 순차적으로 진행하면서 메시지를 보내는데 그 시점을 파악하기 힘든 시퀀스를 디버깅할 때 유용합니다. 문제의 시발점에서 해결의 실마리를 얻을 수 있죠.

Netflix에서 RxJava의 배포전 버전을 위해 제작한 RxJava Debug라는 공식 컴퍼넌트도 있습니다. 업데이트가 되지는 않으며 RxJava-core의 배포전 버전에 기반하고 있습니다. RxJava의 플러그인 API를 사용하고 출력되는 내용이 많습니다. 질문 사항은 Twitter로 보내주세요.

다음: 안드로이드 아키텍처 #4: 클린 아키텍처를 도입해 기본 앱을 전환한 사례를 소개합니다.