RxSwift) Transforming Operater (flatMap, flatMapFirst, flatMapLatest)

flatMap

flatMap은 하나의 Observable이 발행하는 항목들을 여러개의 Observable로 변환하고, 항목들의 배출을 차례차례 줄 세워 하나의 Observable로 전달한다

let disposeBag = DisposeBag()
let numbers = [1,2,3,4,5,6,7,8,9,10]
Observable.from(numbers)
    .flatMap { number -> Observable<String> in
        switch number {
        case 1:
            return Observable.repeatElement("1️⃣").take(3)
        case 2:
            return Observable.repeatElement("2️⃣").take(3)
        case 3:
            return Observable.repeatElement("3️⃣").take(3)
        case 4:
            return Observable.repeatElement("4️⃣").take(3)
        case 5:
            return Observable.repeatElement("5️⃣").take(3)
        default:
            return Observable.just("❌")
        }
    }
    .subscribe { print($0) }
    .disposed(by: disposeBag)

출력

next(1️⃣)
next(1️⃣)
next(2️⃣)
next(1️⃣)
next(2️⃣)
next(3️⃣)
next(2️⃣)
next(3️⃣)
next(4️⃣)
next(3️⃣)
next(4️⃣)
next(5️⃣)
next(❌)
next(4️⃣)
next(5️⃣)
next(❌)
next(5️⃣)
next(❌)
next(❌)
next(❌)
completed

FlatMap 연산자는 옵저버블에서 방출된 값을 변경후 다른 옵저버블로 방출합니다.

이렇게 방출된 옵저버블은 Inner Observable 이라고 불립니다. 내부에서만 처리되는 옵저버블입니다.

최종적으로 나온 결과를 하나의 옵저버블로 재배열하여 방출합니다. 이 과정을 (Flattening : 평탄화)과정이라고 합니다.

이렇게 나온 결과 옵저버블은 말그대로 Result Observable 이라고 합니다.

FlatMap 은 Inner Observable을 통해서 변경후 Result Observable로 방출되면 지연없이 방출됩니다.

이렇게 순차적으로 안들어오는 현상을 Interleaving 이라고 합니다.

flatMapFirst

let disposeBag = DisposeBag()
let numbers = [1,2,3]

func numberChange(number : Int) -> Observable<String> {
    switch number {
    case 1:
        return Observable.repeatElement("🍎").take(3)
    case 2:
        return Observable.repeatElement("🍊").take(3)
    case 3:
        return Observable.repeatElement("🥝").take(3)
    default:
        return Observable.just("❌")
    }
}

let subjset = PublishSubject<Int>()
subjset
    .flatMapFirst { numberChange(number: $0)}
    .subscribe { print($0) }
    .disposed(by: disposeBag)

subjset.onNext(1)
subjset.onNext(2)
subjset.onNext(3)

출력

next(🍎)
next(🍎)
next(🍎)
next(🍊)
next(🍊)
next(🍊)
next(🥝)
next(🥝)
next(🥝)

flatMapfirst 는 flatMap기반으로 파생된 이벤트 입니다.

flatMapfirst은 원본 옵저버블이 방출하는 Next 이벤트를 inner옵저버블로 변환하고

inner옵저버블이 방출하는 모든 결과를 지연 없이 방출합니다.

가장먼저 들어온 inner옵저버블만 사용하고 후에 들어온 옵저버블은 이벤트가 방출되지 않습니다.

flatMapfirst는 처음들어온 옵저버블에 이벤트만 방출하고 그 방출이 끝날떄까지 하나의 주기로 잡습니다.

기존 주기가 끝나고 다음 옵저버블 이벤트가 발생하면 그 이벤트가 first로되어 방출됩니다.

즉, flatMapFirst는 flatMap과 마찬가지로 새로운 Observable을 만들지만,

새로운 옵저버블은 동작이 다 끝날 때 까지 새로 발행된 이벤트를 무시합니다.

flatMapLatest

let disposeBag = DisposeBag()
let numbers = ["1","2","3"]
let apple = "🍎"
let orange = "🍊"
let pear = "🍐"

let psubject = PublishSubject<String>()
let trigger = PublishSubject<Void>()

psubject
    .flatMapLatest { number -> Observable<String> in
        switch number {
        case "1":
            return Observable<Int>.interval(.seconds(1), scheduler: MainScheduler.instance)
                .map { _ in apple}
                .take(until: trigger)
        case "2":
            return Observable<Int>.interval(.seconds(1), scheduler: MainScheduler.instance)
                .map { _ in orange}
                .take(until: trigger)
        case "3":
            return Observable<Int>.interval(.seconds(1), scheduler: MainScheduler.instance)
                .map { _ in pear}
                .take(until: trigger)
        default:
            return Observable.just("")
        }
    }
    .subscribe { print($0) }
    .disposed(by: disposeBag)

psubject.onNext("1")

DispatchQueue.main.asyncAfter(deadline: .now() + 2) {
    psubject.onNext("2")
}

DispatchQueue.main.asyncAfter(deadline: .now() + 4) {
    psubject.onNext("3")
}

DispatchQueue.main.asyncAfter(deadline: .now() + 10) {
    trigger.onNext(())
}

출력

next(🍎)
next(🍎)
next(🍊)
next(🍐)
next(🍐)
next(🍐)
next(🍐)
next(🍐)

flatMapLatest 로 인해 나온 옵저버블 이후 새로운 옵저버블 이벤트가 발생하면 이전 옵저버블은 종료됩니다.

따라서 flatMap처럼 옵저버블에서 동시다발적으로 Next이벤트가 들어와도 기존 이벤트를 내보내던 옵저버블이 종료되어 위와 같은 출력결과가 나옵니다.