RxSwift) Transforming Operators (merge)

merge 연산자는 옵저버블에서 방출되는 여러 이벤트를 하나의 옵저버블에서 방출하도록 만들어줍니다.

concat 연산자와 비슷하지만 다른점은

concat 은 순차적으로 합쳐지는 옵저버블 A, B 가 있다면 A옵저버블에서 발생된 모든 이벤트가 방출된 다음 B옵저버블에서 이벤트가 방출됩니다.

하지만 merge는 A와 B옵저버블이 순차적으로 이벤트를 방출하면 그때마다 이벤트를 받을 수 있습니다.

우선 예제 코드를 보겠습니다.

let disposeBag = DisposeBag()

enum MyError: Error {
   case error
}

let subjectA = BehaviorSubject(value: "🍎")
let subjectB = BehaviorSubject(value: "🍐")
let subjectC = BehaviorSubject(value: "🍊")

let mergeObservable = Observable.of(subjectA, subjectB)

mergeObservable
    .subscribe { print($0) }
    .disposed(by: bag) 

출력

next(RxSwift.BehaviorSubject<Swift.String>)
next(RxSwift.BehaviorSubject<Swift.String>)
completed

이렇게 출력됩니다.

여기서 mergeObservable에 .merge() 연산자를 추가하겠습니다.

mergeObservable
    .merge()
    .subscribe { print($0) }
    .disposed(by: disposeBag)

출력

next(🍎)
next(🍐)

이상황에서 subjectA, subjectB에 순차적으로 이벤트를 발생시켜 보겠습니다.

mergeObservable
    .merge()
    .subscribe { print($0) }
    .disposed(by: disposeBag)

subjectA.onNext("🍎")
subjectB.onNext("🍐")
subjectB.onNext("🍐")
subjectA.onNext("🍎")

출력

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

이렇게 옵저버블 내부 옵저버블을 순차적으로 발생시키는 of 연산자를 통해 subjectA, subjectB를 mergㅊ

그런데 completed 되지 않은것을 확인할 수 있습니다.

merge된 subjectA만 Completed 하겠습니다.

subjectA.onCompleted()

결과는 위 출력결과와 같습니다.

아직 subjectB가 구독중이기에 Completed처리는 되지 않습니다. 하지만 subjectA는 더이상 next 이벤트를 받을 수 없습니다.

subjectB까지 Completed 되어야 완료 할 수 있습니다.

mergeObservable
    .merge()
    .subscribe { print($0) }
    .disposed(by: disposeBag)

subjectA.onNext("🍎")
subjectB.onNext("🍐")
subjectB.onNext("🍐")
subjectA.onNext("🍎")

subjectA.onCompleted()
subjectA.onNext("🍎")

subjectB.onCompleted()

출력

next(🍎)
next(🍐)
next(🍎)
next(🍐)
next(🍐)
next(🍎)
completed

merge(maxConcurrent:)

merge(maxConcurrent:)는 인자의 수만큼 옵저버블의 이벤트를 받을 수 있습니다.

아래와 같이 merge하는 옵저버블은 3개인데 mexConcurrent에는 2개만 merge 하도록하면

subjectC에서 이벤트가 발생해도 next 이벤트를 방출하지 않습니다.

하지만 앞서 merge 된 subjectA, subjectB둘중 하나라도 Completed된다면 subejctC 는 이벤트를 받을 수 있습니다.

let mergeObservable = Observable.of(subjectA, subjectB)
mergeObservable
    .merge(maxConcurrent: 2)
    .subscribe { print($0) }
    .disposed(by: disposeBag)

subjectC.onNext("🍊")
subjectA.onNext("🍎")
subjectA.onCompleted()
subjectC.onNext("🍊")

출력

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

출력결과를 확인해보면

BehaviorSubject 로 만들어진 초기값이 두번 출력되는 것을 확인할 수 있습니다.

1. 🍎
2. 🍐

다음으로 발생항 이벤트인 subjectC.onNext("🍊") 는 무시되고

1. 🍎

그리고 subjectA는 onCompleted 처리 되었습니다.

후에 이미 생성된 BehaviorSubject 초기값인 🍊 가 방출되며

1. 🍊

subjectC.onNext("🍊") 를통한 Next 이벤트가 방출됩니다.

1. 🍊

마지막으로 Error는 merge된 옵저버블중 하나라도 Error이벤트가 발생하면 Error 이벤트를 통해 종료됩니다.