[RxJava] 조합 연산자(Combining Operators)

조합 연산자는 두개의 데이터를 조합해서 새로운 Observable을 만드는 연산자입니다.

combineLatest

combineLatest 마블다이어그램

combineLatest는 두개의 Observable에서 각각 데이터가 생성될 때 데이터를 조합해서 전달하는 연산자입니다.

첫 데이터는 두개의 Observable에서 모두 데이터가 생성됐을 때 전달이 되고, 그 이후에는 각각의 Observable에서 데이터가 생성 될 때 데이터를 조합해서 전달합니다. 예를 들면, Validation Check를 할 때에 사용할 수 있습니다.

fun main() {

    //combineLatest
    val observable3 = Observable.interval(1000L, TimeUnit.MILLISECONDS)
    val observable4 = Observable.interval(750L, TimeUnit.MILLISECONDS)

    val combineObservable = Observable.combineLatest<Long, Long, String> (
        observable3,
        observable4,
        BiFunction<Long, Long, String> { t1, t2 ->
            "$t1 $t2"
        }
    )
    combineObservable.subscribe {
        println(System.currentTimeMillis())
        println(it)
    }
    Thread.sleep(3000)
}

실행결과

combineLatest 실행결과

merge

merge 마블다이어그램

merge는 두개의 Observable을 합쳐서 하나의 Observable로 만들 수 있는 연산자입니다.

fun main() {
	val observable1 = Observable.interval(0, 1000L, TimeUnit.MILLISECONDS).map {"Observable 1 -> $it"}
    val observable2 = Observable.interval(0, 500L, TimeUnit.MILLISECONDS).map {"Observable 2 -> $it"}

    val mergeObservable = Observable.merge(observable1, observable2)

    mergeObservable.subscribe {
        println(System.currentTimeMillis())
        println(it)
    }
    
    /*
    // mergeWith
    observable1.mergeWith(observable2).subscribe {
        println(System.currentTimeMillis())
        println(it)
    }*/
    
    Thread.sleep(3000)   
}

실행결과

merge 실행결과

concat

concat은 merge와 마찬가지로 두개의 Observable을 합치지만 순서를 보장합니다.

여기서 순서를 보장한다는 말은, 앞의 Observable이 종료될 때 까지 뒤의 Observable이 데이터의 방출을 기다리고, 앞의 Observable이 종료되면 뒤의 Observable의 방출을 시작한다는 의미입니다.

fun main() {

	val observable1 = Observable.interval(0, 1000L, TimeUnit.MILLISECONDS).map {"Observable 1 -> $it"}
        .take(3000L, TimeUnit.MILLISECONDS)
    val observable2 = Observable.interval(0, 1000L, TimeUnit.MILLISECONDS).map {"Observable 2 -> $it"}

    Observable.concat(observable1, observable2)
        .doOnNext {
            println("observable: $it")
        }
        .subscribe({

        }, {

        })
    
	/*
    // concatWith
    observable1.concatWith(observable2)
        .doOnNext {
            println("observable: $it")
        }
        .subscribe({

        }, {

        })*/
        
    Thread.sleep(7000L)
}

실행결과

startWith

startWith 마블다이어그램

startWith는 Observable에 첫번째 데이터를 추가하는 연산자입니다.

fun main() {

    //startWith
    Observable.just(1,2,3,4,5)
        .startWith(100)
        .subscribe { println(it)}
}

실행결과

startWith 실행결과

zip

zip 마블다이어그램

 

zip은 각각의 Observable 데이터의 순서에 맞춰서 조합하는 연산자입니다.

두개의 Observable의 데이터가 모두 들어와야 데이터를 방출합니다.

fun main() {

    //zip
    val observable7 = Observable.just(1, 2, 3, 4, 5,6,7)
    val observable8 = Observable.just("a", "b", "c", "d", "e")

    val zipObservable = Observable.zip(
        observable7,
        observable8,
        BiFunction<Int, String, String> { t1, t2 ->
            "$t1 $t2"
        }
    )
    zipObservable.subscribe { println(it)}
}

실행결과

zip 실행결과