RxJava로 기존 Android code 개선하기

nittaku.tistory.com/179

Rxandroid를 쓰는 이유

 

 

1. thread 동기화처리

Async Task

 

2. UI 다루기

클릭의 추상화

observable 병합

Observable 컴바인

 

 

3. Eventbus

Subject 객체는 Observable과 Observer 모두 가능하므로, 데이터를 받고 전달해주는 Eventbus 가 될 수 있다.

 

 

 

 

 

 

4. 데이터 검색

Realm에 저장된 데이터를 가져올때, .asObservable()로 받은다음 rxJava의 .filter()를 쓰면 데이터를 쉽게 걸러낼 수 있고

마지막에 subscrib()만 해주면됨

 

 

5. 다중 api 호출 용이

server ↔ api ↔ client(retrofit+Rxjava)

 

RxJava를 안드로이드에서 활용하는 가장 많은 부분은 네트워크 모듈이다. 

Retrofit2 만을 사용해서 단순한 API 호출의 비동기통신을 1회성 사용하는게 아니라 여러 비동기 API 통신을 처리할수 있다. 

 

1) Retrofit2의 API 통신 Interface 작성시, with RxJava2

 

 

2) Retrofit2의 단일 API를 호출할 때, with RxJava2

만약 하나의 api만 호출할것이라면 굳이 RxJava2를 쓸 필요가 없다. 

 

3) Retrofit2의 다중 API를 호출할 때, with RxJava2

 

에러 핸들링

return observableFactory.<T>toObservable(this)
	 .compose(this.<T>transform(observableRequest))
	 .observeOn(Schedulers.io())
	 .map(new ResponseMetadataOperator<>(this))
	 .flatMap(this::mapResponse)
	 .observeOn(AndroidSchedulers.mainThread())
	 .<AirResponse<T>>compose(group.transform(tag))
	 .doOnError(new ErrorLoggingAction(request))
	 .doOnError(NetworkUtil::checkForExpiredToken)
	 .subscribeOn(Schedulers.io())
	 .subscribe(request.observer());

에러 로그를 위해 doOnError를 사용했습니다. 네트워크 통신이 실패할 때 분석 서비스에 로그를 남기고 몇 번이나 실패가 발생하는지 알고 싶다고 가정해 봅시다. doOnError는 스트림에서 에러를 받을 때마다 실행되는 액션으로 여러 번 발생할 수 있으므로 하나의 스트림에 여러 번의 에러 핸들링을 할 수 있습니다. doOnError가 에러 이벤트를 발견하면 설정된 액션을 호출하는데, 이는 단지 사이드 이펙트에 불과합니다.

return observableRequest
	 .rawRequest()
	 .<Observable<Response<T>>>newCall()
	 .observeOn(Schedulers.io())
	 .unsubscribeOn(Schedulers.io())
	 .flatMap(responseMapper(airRequest))
	 .onErrorResumeNext(errorMapper(airRequest));

catch 블럭처럼 작용하는 onErrorResumeNext를 사용할 수도 있는데 사실 반응형 프로그래밍과는 어울리지 않습니다. 마치 에러를 발견할 때마다 ‘이 에러를 catch하고 예외를 래핑해서 로그하고 빈 데이터를 리턴하는 액션을 여기서 실행해야지.’’ 하는 것과 마찬가지입니다. 명령형 프로그래밍처럼 생각한다면 catch 블럭처럼 작용할 수밖에 없죠

 

유닛테스트

TestSubscriber

비동기 데이터 스트림에서 유닛 테스트를 한다는 것은 복잡한 일이므로 RxJava는 TestSubscriber라는 클래스를 지원합니다.

@Test public void testErrorResponseNonJSON() {
 server.enqueue(new MockResponse()
	 .setBody("something bad happened")
	 .setResponseCode(500));
 TestRequest request = new TestRequest.Builder<String>().build();
 TestSubscriber<AirResponse<String>> subscriber = new TestSubscriber<>();
 observableFactory.<String>toObservable(request).subscribe(subscriber);
 subscriber.awaitTerminalEvent(3L, TimeUnit.SECONDS);
 NetworkException exception = (NetworkException)
	 subscriber.getOnErrorEvents().get(0);
 assertThat(exception.errorResponse(), equalTo(null));
 assertThat(exception.bodyString(), equalTo("something bad happened"));
}

 

TestSubscriber를 사용해서 스트림을 subscribe하고 스레드를 블럭해서 이벤트를 받을 때까지 기다릴 수 있습니다. 예를 들어 .awaitTerminalEvent 메서드는 onCompleted나 onError 등 끝내는 이벤트가 올 때까지 스레드를 블럭합니다. 스트림의 모든 이벤트마다 onNext를 여러번 만나게 되며, 이벤트가 끝나면 성공이나 실패에 따라 onCompleted나 onError를 받습니다. 마침내 더 이상 어떤 이벤트도 받을 수 없게 되면 스트림을 끝내게 되죠.

@Test public void testUnicodeHeader() {
 server.enqueue(new MockResponse().setBody("\"Hello World\""));
 TestRequest request = new TestRequest.Builder<String>()
 .header("Bogus", "中華電信")
 .build();
 observableFactory.toObservable(request)
 .toBlocking()
 .first();
 RecordedRequest recordedRequest = server.takeRequest();
 assertThat(recordedRequest.getHeader("Bogus"), equalTo("????"));
}

스레드를 블럭하는 다른 방법으로 toBlocking을 사용할 수도 있는데, 유닛 테스트에는 좋지만 상용 코드에는 그다지 유용하지 않습니다. RxJava를 사용한다면 스레드를 블럭하는 것을 좋아하지 않겠지만 테스트에는 유용하죠. test subscriber 사용 코드보다 짧고, 언제 실패하지 않는지 알 수 있다면 블럭해서 첫번째 이벤트만을 받을 수 있습니다.

저희는 OkHttp를 사용해서 가상 응답을 만들어서 가짜 응답을 보냈습니다. OkHttp에 버그가 있으면 어떻게 될지 특별한 캐릭터를 헤더를 붙여 테스트를 했습니다. 헤더를 잘 지우는지 테스트한 거죠.

---

넷플릭스에서 RxJava를 만들게된 이유

 

1.  Embrance Concurrency

동시성을 적극적으로 끌어안을 필요가 있다.

자바가 동시성 처리를 하는 데 번거로움이 있다. 이를 해결하려고 넷플릭스는 클라이언트의 요청을 처리하는 서비스 계층에서 동시성을 적극적으로 끌어안았다. 클라이언트의 요청을 처리할 때 다수의 비동기 실행 흐름(스레드 등)을 생성하고 그것의 결과를 취합하여 최종 리턴하는 방식으로 내부 로직을 변경했다.

 

2. Java Futures are Expensive to Compose

자바 Future를 조합하기 어렵다는 점을 해결해야 한다.

2013년 당시 자바 8에서 제공하는 CompletableFuture 같은 클래스가 제공되지 않았기 때문이다. 그래서 비동기 흐름을 조합할 방법이 거의 없었다. RxJava에서는 이를 해결하려고 비동기 흐름을 조합할 수 있는 방법을 제공한다. RxJava에서는 조함하는 실행 단위를 리액티브 연산자라고 한다.

3. Callbacks Have Their Own Problems

콜백 방식의 문제점을 개선해야 한다.

콜백이 콜백을 부르는 콜백 지옥 상황이 코드의 가독성을 떨어뜨리고 문제 발생 시 디버깅을 어렵게 만든다.
비동기 방식으로 동작하는 가장 대표적인 프로그래밍 패턴은 콜백이다. 그래서 RxJava는 콜백을 사용하지 않콜백이 콜백을 부르는 콜백 지옥 상황이 코드의 가독성을 떨어뜨리고 문제 발생 시 디버깅을 어렵게 만든다.

 

 

 

 

---

RxJava 도입시 장점

1. 다양한 비동기처리 가능

• 순차적으로 실행되는 비동기통신을, 동기식 .flatemap()으로 처리
• 복수의 비동기처리를 완료후 결과값 합칠때 .merge()로 처리
• 모든 비동기작업은 Observable<T>로 해결

 

2. 콜백지옥 탈출

unsubscriber

 

3. 다양한 스레드 관리 가능

 

4. UI 이벤트 제어

 

5. Lambda 사용시 코드 깔끔

 

6. 읽기전용 Observable 도 가능

 

 

 

 

 

 

 

 

참고

nittaku.tistory.com/179

taeiim.tistory.com/entry/RxJava2-1-%EB%A6%AC%EC%95%A1%ED%8B%B0%EB%B8%8C-%ED%94%84%EB%A1%9C%EA%B7%B8%EB%9E%98%EB%B0%8D%EA%B3%BC-RxJava

academy.realm.io/kr/posts/compose-everything-rx-kotlin/