Observable을 필터링하는 연산

이번 글에서는 Observable로부터 발행되는 아이템들을 선택적으로 발행하도록 하는 연산자들을 알아보도록 하겠습니다.

debounce 연산자

• debounce 연산자는 특정 시간 동안 다른 아이템이 발행되지 않을 때만 아이템을 발행하도록 하는 연산자입니다.

• debounce 또한 RxJava #2 - 변형 연산자의 buffer 연산과 같이 흐름 제어 연산자입니다. 빠르게 연속 이벤트를 처리할 수 있는 흐름 제어 함수입니다. 예를 들면 버튼을 빠르게 누르는 상황에서 마지막으로 누른 이벤트만을 처리하는 경우와 같이 사용할 수 있습니다.

• 반복적으로 빠르게 발행된 아이템들을 필터링할 때 유용합니다.

• 마블 다이어그램을 보면 첫 번째 시간 간격 동안 빨간색 원 아이템만 발행됐기에 해당 아이템을 방출해주고, 두 번째 시간 간격 동안 노란색과 초록색이 발향됐기에 해당 시간에는 아이템을 방출하지 않는 것을 확인할 수 있습니다.

Observable.create { emitter ->
    emitter.onNext("1")
    Thread.sleep(100)
    emitter.onNext("2")
    emitter.onNext("3")
    emitter.onNext("4")
    emitter.onNext("5")
    Thread.sleep(100)
    emitter.onNext("6")
}.debounce(10, TimeUnit.MILLISECONDS)
    .subscribe { println(it) }

Thread.sleep(300)

/**
 * 결과
 * 1
 * 5
 * 6
 */

distinct

• distinct 연산자는 이미 발행한 아이템을 중복해 발행하지 않도록 필터링합니다.

• 자바의 Set 자료구조와 유사하다고 생각하면 될 것 같습니다.

• 마블 다이어그램을 보면 빨간색 원이 두 번 발행됐지만 중복된 것이기에 한 번만 방출되는 것을 확인할 수 있습니다.

Observable.just(1,2,2,1,3)
    .distinct()
    .subscribe { println(it) }

/**
 * 결과
 * 1
 * 2
 * 3
 */

elementAt

• elementAt 연산자는 발행되는 아이템 시퀀스에서 특정 인덱스에 해당하는 아이템을 필터링합니다.

• 마블 다이어그램을 보면 인자로 주어진 2번째 인덱스의 아이템만 방출하는 것을 확인할 수 있습니다.

Observable.just(1,2,3,4)
    .elementAt(2)
    .subscribe { println(it) }

/**
 * 결과
 * 3
 */

filter 연산자

• 조건식이 true일 때만 아이템을 방출하도록 합니다. 즉, 주어진 술어를 만족하는 아이템만을 넘깁니다.

• Kotlin Colletion 연산자의 filter와 개념이 같습니다.

• 마블 다이어그램을 보면 술어로 원을 주었기에 원 모양의 아이템만 방출된 것을 확인할 수 있습니다.

Observable.just(2,30,22,5,60,1)
    .filter { x -> x > 10 }
    .subscribe { println(it) }

/**
 * 결과
 * 30
 * 22
 * 60
 */

sample

• Sample 연산자는 일정 시간 간격으로 최근에 Observable이 발행한 아이템들을 방출하는 연산자입니다.

• Sample 연산자도 buffer, debounce 연산과 같이 흐름 제어 연산에 해당합니다.

• 해당 시간에 아무리 많은 데이터가 들어와도 해당 구간의 마지막 데이터만이 방출되고 나머지는 무시됩니다.

• 마블 다이어그램을 보면 빨간색, 주항색 원 데이터가 발행되었지만 첫 번째 시간에 가장 마지막 데이터인 주황색 원만 방출된 것을 확인할 수 있습니다.

val disposable = Observable.interval(100, TimeUnit.MILLISECONDS)
    .sample(300, TimeUnit.MILLISECONDS)
    .subscribe {
        println(it)
    }

Thread.sleep(1000)

/*
결과
2
4
7
 */

• 예제 코드는 100 Milliseconds마다 값을 발행하고 300 Milliseconds 마다 가장 마지막에 발행된 아이템을 방출하도록 하고 있습니다. 결과가 (2, 4, 7), (2, 5, 7), (2, 5, 8) 왔다갔다 하는데, 결론적으로는 해당 시점마다 가장 최근에 발행된 데이터만을 방출한다는 것입니다.

skip

• Observable이 발행하는 n개의 아이템을 무시하고 이후에 나오는 아이템을 방출하는 연산자입니다.

• 마블 다이어그램을 보면 인자로 3을 전달하였기에 앞에 세 개의 원 아이템은 무시하고 뒤의 세 개의 원 아이템만 방출한 것을 확인할 수 있습니다.

Observable.just(1,2,3,4)
    .skip(2)
    .subscribe { println(it) }

/**
 * 결과
 * 3
 * 4
 */

take

• take 연산자는 skip 연산자와 반대로 Observable이 처음 발행하는 n개의 아이템만 방출하도록 하는 연산자입니다.

• n번째 뒤로 발행되는 아이템은 모두 무시합니다.

• 마블 다이어그램을 보면 인자로 2를 전달하였기에 2개의 아이템만을 방출하고, 나머지 아이템은 무시하고 있습니다.

Observable.just(1,2,3,4)
    .take(2)
    .subscribe { useTimber("Data : $it") }
    
/**
 * 결과
 * 1
 * 2
 */

all 연산자

• 모든 발행되는 아이템이 특정 조건을 만족할 때 true를 반환합니다.

• 만약 아이템 중 하나라도 조건에 부합하지 않는다면 false를 반환합니다.

• Kotlin Collection 함수의 all 연산자와 똑같은 기능을 수행합니다.

• 마블 다이어그램을 보면 첫 번째 Observable이 발행하는 모든 아이템은 5보다 작기에 true를 방출하였지만, 두 번째 Observable이 발행하는 아이템 중에 7은 5보다 작다는 술어를 만족하지 못하기에 false를 방출하는 것을 확인할 수 있습니다.

Observable.just(1,2)
    .all { num -> num > 0 }
    .subscribe(::println)

/**
 * 결과
 * true
 */

Observable.just(0,1,2)
    .all { num -> num > 0 }
    .subscribe(::println)

/**
 * 결과
 * False
 */

amb 연산자

• 여러 개의 Observable을 동시에 구독하고, 그중 가장 먼저 아이템을 발행하는 Observable을 선택하고 싶다면 amb 연산자를 사용하면 됩니다.

• 마블 다이어그램을 보면 두 개의 Observable이 있는데, 첫 번째 Observable이 먼저 원 아이템을 발행하기에 해당 Observable의 아이템만 방출하고 두 번째 Observable은 무시하는 것을 확인할 수 있습니다.

val list = ArrayList<Observable<Int>>()
list.add(Observable.just(20,40,60).delay(100, TimeUnit.MILLISECONDS))
list.add(Observable.just(1,2,3))
list.add(Observable.just(0,0,0).delay(200, TimeUnit.MILLISECONDS))
Observable.amb(list).subscribe { println(it) }

/**
 * 결과
 * 1
 * 2
 * 3
 */

• 예제 코드에서 (20, 40, 60)을 가진 Observable과 (0, 0, 0)을 가진 Observable은 delay를 하기에 (1, 2, 3)을 가진 Observable이 가장 먼저 데이터를 발행합니다. 따라서 (1, 2, 3)을 가진 Observable이 발행하는 아이템만 방출하고 나머지 Observable은 무시하고 있습니다.

참고 및 참조
틀린 부분은 댓글로 남겨주시면 수정하겠습니다..!!
2022-09-23에 작성되었습니다.

아키텍처를 알아야 앱 개발이 보인다
RxJava Docs