테스트 케이스로 작성해보는 RxJava 함수

이제는 코루틴으로 인해, 사용빈도가 적어지는 자주 사용했던 Rx 함수들을 정리하는 시간을 가지려고합니다. 그리고 안녕...

 

제대로 동작하는 것을 확인하기 위해, 테스크 케이스로 작성하고 딱 20개만 해보려고합니다.

 

20 Test passed!

먼저 전체 코드를 링크 걸어두겠습니다

 

Single, Completable, Observable, Flowable, Maybe 의 목적과 특성은 이미 알고 있을 것이라고 가정하고,

위의 자료형과 상관없이 생각나는 함수 위주로 작성해보겠습니다.

 

1. Just 함수

/**
 * Single.just, SingleJust : Single 로 Wrapping 된 아이템을 한번 방출한다.
 * 특정 아이템을 Single 타입으로 변경하여 체이닝을 이어갈 때 사용하면 유용하겠다.
 * ex) flatMap 으로 벗겨진 타입을 다시 Single 로 감싸준다
 */
@Test
fun `Single just - emit one number`() {
    Single.just(7).subscribe { seven -> assertThat(seven).isEqualTo(7) }
            .addTo(compositeDisposable)
}

@Test
fun `SingleJust - emit one number`() {
    SingleJust(15).subscribe { seven -> assertThat(seven).isEqualTo(15) }
            .addTo(compositeDisposable)
}

@ParameterizedTest
@MethodSource(value = ["provideInteger"])
fun `Single just - emit multiple number & reduce`(numbers: Single<Array<Int>>, result: Int) {
    numbers.subscribe { number ->
        val actual = number.reduce { acc, i -> acc + i }
        assertThat(actual).isEqualTo(result)
    }.addTo(compositeDisposable)
}

 

2. Delay 함수

/**
 * Delay : 해당 시간 이후 아이템을 배출
 * 특정 비즈니스 로직을 수행하기 위해 사용되는 경우가 많다.
 * ex) 아이템을 보여주고, 500 ms 뒤에는 사라지게 만든다
 */
@Test
fun `Single delay`() {
    val testObserver = SingleJust(listOf(2, 4))
            .delay(500L, TimeUnit.MILLISECONDS).test()

    testObserver.assertEmpty()

    testScheduler.advanceTimeBy(500L, TimeUnit.MILLISECONDS)

    testObserver.assertValue(listOf(2, 4))
}

 

3. AmbArray 함수

/**
 * AmbArray : 가장 먼저 관찰된 아이템만 배출하고, 이후 나머지는 모두 무시한다
 * ex) Gps_provider, Network_provider 등을 사용하여 사용자의 좌표를 받아오는데, 가장 먼저 관측된 아이템만 사용한다
 */
@Test
fun `Single ambArray`() {
    val item1 = SingleJust(listOf(1, 2))
            .delay(500L, TimeUnit.MILLISECONDS)
    val item2 = SingleJust.just(listOf(3, 4))
            .delay(100L, TimeUnit.MILLISECONDS)
    val item3 = SingleJust.just(listOf(5, 6))
            .delay(900L, TimeUnit.MILLISECONDS)

    val testObserver = Single.ambArray(item1, item2, item3).test()

    testScheduler.advanceTimeBy(2L, TimeUnit.SECONDS)

    testObserver.assertValue(listOf(3, 4))
}

 

4.Timer 함수

/**
 *  Timer : 일정시간 이후 1회 배출 되며 이때, Long 타입 0을 반환합니다
 *  A 이후 정해진 시간 이후 B 가 호출되어야하는 상황에 트리거의 역할을 할 수 있습니다.
 */
@Test
fun `Single timer`() {
    val testObserver = Single.timer(500L, TimeUnit.MILLISECONDS).test()
    testScheduler.advanceTimeBy(1L, TimeUnit.SECONDS)
    assertThat(testObserver.values()).isEqualTo(listOf(0L))
}

 

5. Map 함수

/**
 * Map : 전달 받은 데이터를 새로운 형태로 변형시켜준다
 * ex) 특정 자료형에 공통된 변경을 적용할 때 매우 많이 사용됨
 */
@Test
fun `Single map - one item`() {
    Single.just("Lucas").map { "$it.shin" }.subscribe { it ->
        assertThat(it).isEqualTo("Lucas.shin")
    }
}

@ParameterizedTest
@MethodSource("provideMapMock")
fun `Single map - multiple item`(list: Single<List<Int>>, expected: List<Int>) {
    list.map { ints ->
        ints.map { int ->
            int * 11
        }
    }.subscribe { ints ->
        assertThat(ints).isEqualTo(expected)
    }
}

 

6. FlatMap 함수

/**
 * FlatMap : Map 과 마찬가지로 Single<R> 을 리턴하지만, 파리머터에 SingleSource 타입의 자료형이 반환되어야 합니다.
 */
@ParameterizedTest
@MethodSource("provideFlatMapMock")
fun `Single flatMap`(list: Single<List<Int>>, expected: List<Int>) {
    // Function<? super T, ? extends SingleSource<? extends R>> mapper
    val flatMapList = list.flatMap { ints ->
        val newFlatMapList = ints.map { int -> int.toString() }
        Single.just(newFlatMapList)
    }.blockingGet()

    // Function<? super T, ? extends R> mapper
    val mapList = list.map { ints ->
        ints.map { int -> int.toString() }
    }.blockingGet()

    assertThat(flatMapList).isEqualTo(mapList)
}

 

7. Concat 함수

/**
 * Concat : 아이템들을 모아서 호출된 순서에 맞게 방출합니다
 * Rx Operator 에서 concat 이라는 키워드가 들어가있으면 순서를 보장해준다고 생각하면 편합니다
 * 아래 예제에서 listOf(3, 4) 의 배출이 더 빠를 것 같지만, concat 메서드의 파리머터로 item1 이 먼저 호출되었기 때문에
 * 500L 이후에 list(1, 2) 이 배출되고 이후에 100L 뒤에 list(3, 4) 가 호출됩니다. 따라서 앞의 메서드에서 오버헤드가 큰
 * 일을 하게되면 성능에 문제가 될 수 있습니다
 * ex)처리 순서가 중요할 때 사용이 용이합니다
 */
@Test
fun `Single concat`() {
    val item1 = SingleJust(listOf(1, 2)).delay(500L, TimeUnit.MILLISECONDS)
    val item2 = SingleJust(listOf(3, 4)).delay(100L, TimeUnit.MILLISECONDS)
    val item3 = SingleJust(listOf(5, 6)).delay(900L, TimeUnit.MILLISECONDS)

    val testObserver = Single.concat(item1, item2, item3).test()

    testObserver.assertEmpty()

    testScheduler.advanceTimeBy(500L, TimeUnit.MILLISECONDS)

    testObserver.assertValue(listOf(1, 2))

    testScheduler.advanceTimeBy(100L, TimeUnit.MILLISECONDS)

    testObserver.assertValues(listOf(1, 2), listOf(3, 4))

    testScheduler.advanceTimeBy(900L, TimeUnit.MILLISECONDS)

    testObserver.assertValues(listOf(1, 2), listOf(3, 4), listOf(5, 6))
}

 

8. fromXXXX 함수

/**
 * FromCallable, FromObservable, FromPublisher, FromFuture : 각 from prefix 를 제외한 타입을 파라미터로 받는다
 * 비동기 처리 이후 실행 결과를 리턴할 수 있다. 비동기 처리 이후 결과를 받아 또 다른 로직을 실행해야 하는 경우에 유용하다
 * ex) 내부 DB 에서 로직 수행 후 이후 결과를 받아, 다른 처리에 사용
 */
@Test
fun `Single - fromXXXXX`() {
    fun emitList() = listOf(1, 2, 3, 4, 5)

    // fromCallable
    val fromCallable1 = Single.fromCallable(Callable(::emitList)).blockingGet()
    val fromCallable2 = Single.fromCallable { emitList() }.blockingGet()

    assertThat(fromCallable1).isEqualTo(fromCallable2)

    // fromObservable
    val fromObservable = Single.fromObservable(Observable.just(listOf(1, 2, 3, 4, 5))).test()

    testScheduler.advanceTimeBy(1L, TimeUnit.SECONDS)

    fromObservable.assertValue(listOf(1, 2, 3, 4, 5))

    // fromPublisher
    // Processor  backPressure 지원하는 Subject
    // Subject 에는 PublishSubject, BehaviorSubject(latest), ReplaySubject(cache),
    // UnicastSubject(emit after subscribe) 등이 있고 각각 특징이 있어서 상황에 맞게 잘 골라 사용해야한다
    val fromPublisher = Single.fromPublisher(PublishProcessor.just(100)).test()
    fromPublisher.assertValue(100)

    // fromFuture
    val futureTask = Executors.newSingleThreadExecutor().submit<List<Int>> { listOf(1, 2, 3) }
    val fromFuture = Single.fromFuture(futureTask).test()
    fromFuture.assertValue(listOf(1, 2, 3))
}

 

9. Interval & filter

/**
 * Flowable : Single 이 하나의 아이템을 배출하면 Flowable 은 여러개 배출가능(그렇게 때문에 backPresure policy 존재)
 * Interval : 정해진 간격으로 아이템 배출
 * Filter : 작성한 조건의 결과가 true 결과면 아이템은 배출합니다
 * Take : 원하는 횟수만큼 아이템을 배출한다. (만약 배출하려는 타겟의 크기가 횟수보다 작게되면 모두 방출하고 완료(onComplete) 호출)
 */
@Test
fun `Flowable - interval & filter`() {
    val testObserver = Flowable.interval(1L, TimeUnit.SECONDS)
            .filter { int -> int.toInt() % 2 == 0 }.take(10).test()

    testScheduler.advanceTimeBy(10L, TimeUnit.SECONDS)

    testObserver.assertValues(0, 2, 4, 6, 8)
}

 

10. Distinct 함수

/**
 * Distinct : 중복되는 아이템을 제거해준다 (내부적으로 createHashSet 로 관리됨)
 */
@Test
fun `Flowable - distinct`() {
    val testObserver = Flowable.just(0, 1, 0, 1, 2, 2)
            .distinct().test()

    testScheduler.advanceTimeBy(10L, TimeUnit.SECONDS)

    testObserver.assertValues(0, 1, 2)
}

 

11. Skip 함수

/**
 * Skip : 지정한 횟수만큼 아이템 배출을 생략한다
 */
@Test
fun `Flowable - interval & skip`() {
    val testObserver = Flowable.interval(1L, TimeUnit.SECONDS)
            .take(10)
            .skip(2)
            .test()

    testScheduler.advanceTimeBy(10L, TimeUnit.SECONDS)

    // Skip 0, 1
    testObserver.assertValues(2, 3, 4, 5, 6, 7, 8, 9)
}

 

12. Zip 함수

/**
 * Zip : 자주 사용되는 Operator 로 배출되는 아이템을 모아서 하나의 아이템으로 모아준다. 최대 9개의 인자를 받을 수 있고,
 * 그 이후로는 zipArray 를 사용해야한다 
 * ex) 불완전한 인자들을 모아 하나의 인자로 만들 때 매우 유용하다.
 */
@Test
fun `Single - zip`() {
    val item1 = Single.just("Nanamare")
    val item2 = Single.just("Kinamare")
    Single.zip(item1, item2, { item1, item2 ->
        val finalItem = "$item1 is $item2"
        assertThat(finalItem).isEqualTo("Nanamare is Kinamare")
    })
}

 

13. Sample 함수

/**
 * Sample : 흐름 제어 Operator 로 배출되는 아이템에서 정해진 간격에서 가장 최근 값들을 관찰한다
 */
@Test
fun `Flowable - sample`() {
    val testSubscriber = Flowable.interval(50L, TimeUnit.MILLISECONDS)
            .sample(100L, TimeUnit.MILLISECONDS, false)
            .take(5)
            .test()

    testScheduler.advanceTimeBy(10L, TimeUnit.SECONDS)

    testSubscriber.assertValues(0, 2, 4, 6, 8)
}

 

14. Debounce 함수

/**
 * Debounce : 흐름제어 Operator 로 이벤트들을 그룹핑하여, 특정 주기의 마지막 값만 배출한다
 * ex) 관련 검색 기능을 사용할 때 유용하다. 가령 Hello 를 입력하면
 *     H -> API 호출 -> e -> API 호출 -> l -> API 호출... 순으로 될 수 있지만,
 *     H -> e -> l -> l -> O -> API 호출로 처리할 수 있다.
 */
@Test
fun `Flowable - Debounce`() {
    val testObserver = Flowable.concat(
            Flowable.timer(100L, TimeUnit.MILLISECONDS).map { 1 },
            Flowable.timer(100L, TimeUnit.MILLISECONDS).map { 2 },
            Flowable.timer(100L, TimeUnit.MILLISECONDS).map { 3 },
            Flowable.timer(100L, TimeUnit.MILLISECONDS).map { 4 },
    ).debounce(300L, TimeUnit.MILLISECONDS).test()

    testScheduler.advanceTimeBy(1L, TimeUnit.SECONDS)

    testObserver.assertValue(4)
}

 

이외에도 자주 사용했던 함수들은

combineLatest, fromIterable, empty, error, complete, create 등이 있네요 :) 

 

여러분이 자주 사용하는 함수가 있다면 코멘트 달아주세요. 추가해보겠습니다!