안드로이드 비동기 처리 2-3 (Operators : 연산자들) 코드편

1. 생성 연산자 

1-1) just() : 인자를 받는 데이터를 순서대로 발행하는 Observable 생성 연산자이며, 최대 10개 까지 발행할 수 있다.

fun operatorJust() {
    val source = Observable.just("1", "2", "3", "4", "5")
    var str = ""

    // 데이터를 소비하는 Observer를 직접 만들수 있지만 그것을 생략할 수도 있다.
    val result = source.observeOn(AndroidSchedulers.mainThread()) // 결과는 메인 스레드에서 관찰
        .subscribe(
            { // onNext
                str = "$str $it"
                _rxJustLiveData.value = "$str"
                Log.d("RxJavaOperatorsViewModel", "operatorJust : $it")
            },
            { error -> error.printStackTrace() }, // onError: 에러 처리
            { _rxJustLiveData.value = "$str Complete" } // onComplete
        )
}

 

1-2) create() : emitter를 이용하여 데이터를 발행하고, onNext(), onComplete(), onError() 함수를 이용하여 데이터를 발행하거나 완료처리, 에러처리

fun operatorCreate() {
    val userData = listOf("A - 20 years old", "B - 32 years old", "C - 40 years old")
    val source = Observable.create { emitter ->
        userData.forEachIndexed { idx, c ->
            emitter.onNext("$idx : $c")
            Thread.sleep(1000) // 1초 지연
        }
        emitter.onComplete()
    }
    val result = source
        .subscribeOn(Schedulers.io()) // 데이터 생성은 IO 스레드에서 실행
        .observeOn(AndroidSchedulers.mainThread()) // 결과는 메인 스레드에서 관찰
        .subscribe {
            _rxCreateLiveData.value = it
        }
}

 

1-3) interval() : 시간 간격을 두고 데이터를 전달하는 stream을 만든다. 따로 지정해주지 않는다면 기본적으로 SchedulerSupport.COMPUTATION 스케쥴러로 별도의 스레드에서 실행 된다.

fun operatorInterval() {
    val source =
        Observable.interval(1000L, TimeUnit.MILLISECONDS).map { data -> (data + 1) }.take(5)
    val result = source.observeOn(AndroidSchedulers.mainThread()) // 결과는 메인 스레드에서 관찰
        .subscribe {
            _rxIntervalLiveData.value = "$it"
        }
}

2. 변환연산자 

2-1) map() : 

fun operatorMap() {
    val numberList = arrayOf("1", "2", "3", "4", "5")
    val source = Observable.fromArray(numberList).map { "number $it" }

    val result = source.observeOn(AndroidSchedulers.mainThread()) // 결과는 메인 스레드에서 관찰
        .subscribe {
            _rxMapLiveData.value = "$it"
        }
}

3. 필터연산자

3-1) debounce() : 이벤트를 그룹화하여 특정시간이 지난 후 하나의 이벤트만 발생하도록 하는 기술이다.

fun operatorDebounce() {
    val numberList = arrayOf("1", "2", "3", "4", "5")

    val result = Observable.fromArray(*numberList) // 배열의 값을 하나씩 방출
        .concatMap { item ->
            // 각 아이템을 Observable로 감싸고 1초 지연
            Observable.just(item).delay(1001, TimeUnit.MILLISECONDS)
        }
        .debounce(1000, TimeUnit.MILLISECONDS) // 1초 동안 이벤트가 없으면 해당 아이템을 방출
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(
            { item ->
                _rxDebounceLiveData.postValue(item) // 방출된 값을 LiveData에 전달
            },
            { error ->
                Log.e("operatorDebounce", "Error: ${error.message}")
            }
        )
}

 

3-2) throttleFirst() : 이벤트를 일정한 주기마다 발생하도록 하는 기술이다.

fun operatorThrottle() {
    val numberList = arrayOf("1", "2", "3", "4", "5", "6", "7", "8", "9", "10")
    val source = Observable.create { emitter ->
        numberList.forEach {
            emitter.onNext(it)
            Thread.sleep(1000) // 1초 지연
        }
        emitter.onComplete()
    }
    val result = source
        .throttleFirst(2, TimeUnit.SECONDS) // 1초 간격으로 최신 이벤트만 방출
        .subscribeOn(Schedulers.io()) // 데이터 생성은 IO 스레드에서 실행
        .observeOn(AndroidSchedulers.mainThread()) // 결과는 메인 스레드에서 관찰
        .subscribe {
            _rxThrottleLiveData.value = it
        }
}