조합 연산자는 두개의 데이터를 조합해서 새로운 Observable을 만드는 연산자입니다.
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)
}실행결과
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)
}실행결과
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는 Observable에 첫번째 데이터를 추가하는 연산자입니다.
fun main() {
//startWith
Observable.just(1,2,3,4,5)
.startWith(100)
.subscribe { println(it)}
}실행결과
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)}
}실행결과
