[RxJava] 02. RxJava Open Source

지난 'Basic Concepts for RxJava'에 이어서, 실제 RxJava를 어떻게 사용하는 보려고 한다.

 

이번에 확인할 Open Source는 Retroit / RxJava, RxAndroid 등을 활용한, Stock Tracker 애플리케이션이다.

github.com/sakshamdhawan22/Android-MVP-RX-DEMO

sakshamdhawan22/Android-MVP-RX-DEMO

우선 위 오픈 소스를 보기 전에, RxJava에서 사용되는 기본 Operator를 집고 넘어가야겠다.

 

1. Map 

 - Observable이 배출한 항목에 함수를 적용한다.

 

2. FlatMap

 - 하나의 Observable이 발행하는 항목들을 여러 개의 Observable로 변환하고, 항목들의 배출을 차례차례 줄 세워 하나의 Observable로 전달한다.

 

위 두 가지의 차이점이 뭔지? 어떤 상황에서 사용하는 것인지? 처음에는 명확한 설명을 찾다가 답을 얻지 못했다.

아래 내용이 정답에 가까운 것 같아 인용한다.

 

"FlatMap behaves very much like map, the difference is that the function it applies returns an observable itself, so it's perfectly suited to map over asynchronous operations."

 

(stackoverflow.com/questions/22847105/when-do-you-use-map-vs-flatmap-in-rxjava)

 

---

이번에는 오픈 소스의 Presenter에서 RxJava를 이용해서, 서버 통신 구현부를 확인해보자.

public void fetchstock(String stockName, boolean inBackground, boolean forceRefresh) {
    if (fetchSubsciption != null) {
        compositeSubscription.remove(fetchSubsciption);
    }
    if (!inBackground && stockView != null)
        stockView.showLoading();

    fetchSubsciption = Observable.interval(15, TimeUnit.SECONDS)
            .startWith(0l)
            .flatMap(aLong -> stockDataRepository.getStock(stockName, forceRefresh))
            .observeOn(AndroidSchedulers.mainThread(), true)
            .subscribe(stock -> {
                if (stockView != null) {
                    stockView.onStockFetched(stock);
                    if (stock != null) {
                        if (stock.getSource().equals(Stock.Source.REMOTE) && stock.isClosed())
                            fetchSubsciption.unsubscribe();
                    }
                }
            }, e -> {
                e.printStackTrace();
                if (stockView != null) {
                    if (e instanceof IOException && !NetworkUtils.isInternetOn(StockTrackerApp.getContext())) {
                        stockView.showError(StringUtils.getString(R.string.no_internet));
                    } else
                        stockView.showError(StringUtils.getString(R.string.something_went_wrong));
                }
            });
            
    compositeSubscription.add(fetchSubsciption);
}

1) Observable.interval(15, TimeUnit.SECONDS)

 - 특정 시간마다, Observable 방출 (15 seconds)

2) flatMap

 - flatMap 내에서, asynchronous 통신 (stockDataRepository.getStock())

3) observeOn(AndroidSchedulers.mainThread(), true)

 - 처리된 결과를 구독자에게 전달하는 스레드를 지정 (MainThread)

 - 호출 시점 하위 스레드를 설정

 - 여러 번 호출할 수 있으며, 호출되면 그다음부터 동작하는 스레드를 변경할 수 있다.

 

* 참고

4) subscribeOn() 

 - Observable에서 구독자가 subscribe() 함수를 호출했을 때, 데이터 흐름을 발행하는 스레드를 지정

 - 호출 시점 상위에 해당하는 부분의 스레드 설정

 - 한번 호출했을 때, 결정한 스레드를 고정하며 이후에는 다시 호출해도 스레드가 변경되지 않는다.

 

---

다음은 flatMap 내, stockDataRepository.getStock() 메서드이다.

@Override
public Observable<Stock> getStock(String stockName, boolean forceRefresh) {
    if (forceRefresh)
        return mRemoteStockDataSource.getStock(stockName, forceRefresh)
                .doOnNext(stock -> cacheStock(stock));
    return Observable.concat(mLocalStockDataSource.getStock(stockName, forceRefresh),
            mRemoteStockDataSource.getStock(stockName, forceRefresh))
            .doOnNext(stock -> cacheStock(stock));

}

1) Observable.concat()

* 참고. concat()과 merge() 비교를 통한 정리

 - concat() : 하나의 Observable이 끝난 뒤, 그 뒤로 이어져서 하나의 Observable로 flat 하게 된다. (순서를 보장한다.)

 - merge() : 두 개의 Observable 값이 하나의 Observable로 합쳐지게 된다. 

private fun exampleConcat() {
    val observerA = Observable.interval(1, TimeUnit.SECONDS)
        .take(5)
        .map { data ->
            data
        }

    val observerB = Observable.interval(1, TimeUnit.SECONDS)
        .take(5)
        .map { data ->
            data * 2
        }

    Observable.concat(observerA, observerB)
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe { data ->
            Log.d("seosh", "$data")
        }
}

// 출력
11-25 22:58:22.682 17581-17581/com.seosh.test D/seosh: 0
11-25 22:58:23.681 17581-17581/com.seosh.test D/seosh: 1
11-25 22:58:24.682 17581-17581/com.seosh.test D/seosh: 2
11-25 22:58:25.681 17581-17581/com.seosh.test D/seosh: 3
11-25 22:58:26.681 17581-17581/com.seosh.test D/seosh: 4

11-25 22:58:27.683 17581-17581/com.seosh.test D/seosh: 0
11-25 22:58:28.682 17581-17581/com.seosh.test D/seosh: 2
11-25 22:58:29.683 17581-17581/com.seosh.test D/seosh: 4
11-25 22:58:30.683 17581-17581/com.seosh.test D/seosh: 6
11-25 22:58:31.683 17581-17581/com.seosh.test D/seosh: 8

 

2) Observable.doOnNext()

 - Observable 아이템을 발행할 때 호출되는 콜백 함수를 등록할 수 있다.

private fun exampleDoOnNext() {
    Observable.interval(1, TimeUnit.SECONDS)
        .take(5)
        .doOnNext { data ->
            Log.d("seosh", "doOnNext : $data")
        }
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe { data ->
            Log.d("seosh", "$data")
        }
}

// 출력
11-25 23:29:32.531 18310-18334/com.seosh.test D/seosh: doOnNext : 0
11-25 23:29:32.532 18310-18310/com.seosh.test D/seosh: 0
11-25 23:29:33.532 18310-18334/com.seosh.test D/seosh: doOnNext : 1
11-25 23:29:33.533 18310-18310/com.seosh.test D/seosh: 1
11-25 23:29:34.531 18310-18334/com.seosh.test D/seosh: doOnNext : 2
11-25 23:29:34.531 18310-18310/com.seosh.test D/seosh: 2
11-25 23:29:35.532 18310-18334/com.seosh.test D/seosh: doOnNext : 3
11-25 23:29:35.532 18310-18310/com.seosh.test D/seosh: 3
11-25 23:29:36.535 18310-18334/com.seosh.test D/seosh: doOnNext : 4
11-25 23:29:36.535 18310-18310/com.seosh.test D/seosh: 4

 

---

마지막으로, Retrofit을 활용한 getStock() 메서드이다.

@Override
public Observable<Stock> getStock(String stockName, boolean forceRefresh) {
    return RetrofitService.getInstance().getStockData(RestApi.FUNCTION.TIME_SERIES_DAILY, stockName, null, null, API_KEY)
            .subscribeOn(Schedulers.io())
            .map(stockApiResponse -> getStockFromStockApiResponse(stockApiResponse));
}

1) subscribeOn()

 - 위에서 언급한 바와 같이, Observable에서 구독자가 subscribe() 함수를 호출했을 때, 데이터 흐름을 발행하는 스레드를 지정

 

 

이상 오픈 소스에서 RxJava와 Retrofit을 활용한, 서버 통신 예제를 확인해보았다.

내일부터는 실제 회사 업무에 'RxJava'를 적용해봐야겠다. :)