flatmap

flatMap(Function<? super T, ? extends ObservableSource<? extends R>> mapper)

• 인자로 전달 된 mapper 에서 반환 된 Observable들에서 배포 되는 데이타를 배포 한다.

Observable flatMap(Function<? super T, ? extends ObservableSource<? extends R>> mapper, boolean delayErrors)

• delayErrors 가 tru 이면 모든 스트림과 에러가 종료 될 때까지 지연 됩니다. false 의 경우 최초의 에러 신호는 그 즉시 스트림을 종료합니다.
• delayErrors 인자의 기본값은 false 입니다.
• 실제 동작에 대한 검토가 필요합니다.

flatMap(Function<? super T, ? extends ObservableSource<? extends R>> mapper, int maxConcurrency)

• 동시에 발행 가능한 Observable 의 갯수를 추가 옵션으로 받습니다.

import io.reactivex.Observable
import java.util.concurrent.TimeUnit

fun flatMapWithInt(obser: Observable<Int>, maxConcurrency: Int): Observable<Int>? {
    val take = Observable.interval(1L, TimeUnit.MILLISECONDS).take(5)
    val flatMap = take.flatMap ({ r -> Observable.interval(2L, TimeUnit.MILLISECONDS).take(10).map { j -> "$r-$j" } },maxConcurrency)
    flatMap.subscribe { println(it) }
    return null
}
fun main(ar : Array<String>){
    flatMapWithInt(obser,1)
    Thread.sleep(10000)
}

• maxConcurrency 를 0을 넣으면 java.lang.IllegalArgumentException: maxConcurrency > 0 required but it was 0 익셉션이 발생합니다.
• 위의 예제에서 1을 넣으면 0-0 부터 0-9까지 배포되고 나서야 1-0이 배포 되며. 2를 넣으면 0-N 과 1-N 이 뒤석여서 전부 배포 된 다음에야 2-0이 배포 되는 것을 볼 수 있습니다.
• 이러한 특성은 flatMap이 기본적으로 각각의 반환된 Observable을 순서와 상관없이 동시에 처리 하기 때문이며. 순서대로 배포 하는 것을 원하면 concatMap을 사용하면 됩니다.