RxSwift - 여러 개의 Observable 을 합치는 방법(startWith, concat, merge, zip)

startWith

 

현재 위치, 네트워크 연결 상태 등 현재 상태, 초기값이 필요한 상황이 있다. 이때 이를 맨 앞에 붙일 수 있다. 

print("-----startWith------")
let zoo = Observable<String>.of("🐳", "🐕", "🦢")

zoo
    .enumerated()
    .map { (index, element) in
        return "아기 " + element + " \(index)"
    }
    .startWith("🧑‍🌾사육사") // String - Observable 로 방출되는 값과 동일한 타입의 값이 들어가야 한다.
    .subscribe(onNext: {
        print($0)
    })
    .disposed(by: disposeBag)
    
-----startWith------
🧑‍🌾사육사
애기 🐳 0
애기 🐕 1
애기 🦢 2

 

① startWith 에 들어가는 값은 Observable 의 onNext 이벤트로 방출되는 value 와 같은 타입이여야 한다!

 

② startWith 의 선언 위치는 중요하지 않다. startWith 안에 있는 값이 먼저 나오고, 순차적으로 값이 방출된다.

 

concat

 

print("-----concat-----")
let animals = Observable<String>.of("🐳", "🐕", "🦢")
let 사육사 = Observable<String>.of("🧑‍🌾사육사")

let 줄서기 = Observable
    .concat([사육사 , animals])

줄서기.subscribe(onNext: {
    print($0)
})
.disposed(by: disposeBag)


-----concat-----
🧑‍🌾사육사
🐳
🐕
🦢

 

다음처럼 바로 해당 Observable Sequence 에 붙일 수 있다.

print("-----concat2-----")
사육사.concat(animals)
    .subscribe(onNext: {
        print($0)
    })
    .disposed(by: disposeBag)
  
  
-----concat2-----
🧑‍🌾사육사
🐳
🐕
🦢

 

concatMap

 

각각의 Sequence 가 다음 Sequence 가 구독되기 전에 합쳐지는 것을 보증한다.

print("-----concatMap-----")
let 어린이집 = [
    "노랑반": Observable.of("👧🏼", "🧒🏻", "👦🏽"),
    "파랑반": Observable.of("👶🏾", "👶🏻")
]

Observable.of("노랑반", "파랑반")
    .concatMap { 반 in
        어린이집[반] ?? .empty()
    }
    .subscribe(onNext: {
        print($0)
    })
    .disposed(by: disposeBag)
    

-----concatMap-----
👧🏼
🧒🏻
👦🏽
👶🏾
👶🏻

 

두 개의 Sequence 를 어떻게 Append 하는지를 나타낸 것이 concatMap 이다.

 

merge

 

순서를 전혀 보장하지 않고 두 Observable 을 합치는 방식이다.

합쳐진 Observable 내부에 있는 강북, 강남은 어떠한 관계를 가지지 않으므로, 하나의 error 만 포함되어도 에러가 방출된다.

print("----------merge1----------")
let 강북 = Observable.from(["강북구", "성북구", "동대문구", "종로구"])
let 강남 = Observable.from(["강남구", "강동구", "영등포구", "양천구"])

Observable.of(강북, 강남)
    .merge()
    .subscribe(onNext: {
        print("서울특별시의 구:", $0)
    })
    .disposed(by: disposeBag)
    

----------merge1----------
서울특별시의 구: 강북구
서울특별시의 구: 성북구
서울특별시의 구: 강남구
서울특별시의 구: 동대문구
서울특별시의 구: 강동구
서울특별시의 구: 종로구
서울특별시의 구: 영등포구
서울특별시의 구: 양천구

 

마치 순서를 보장하는 것 처럼 보이지만, 다르다. 

maxConcurrent 는 merge 를 통해 한번에 받아낼 Observable 의 수를 의미한다. 

예를 들어, maxConcurrent : 1 로 되어 있는데, 강남 이나 강북 중 임의로 하나의 Observable 을 선택하여 onNext 이벤트를 통해 값을 다 방출한 후에, 다음 Observable 의 값을 방출한다는 의미이다.

 

네트워크 요청이 많아질 때, 리소스를 제한하거나 연결 수를 제한하기 위해 maxConcurrent 를 사용할 수는 있으나 자주 사용되지는 않을 연산자이다.

print("----------merge2----------")
Observable.of(강남, 강북)
    .merge(maxConcurrent: 1)
    .subscribe(onNext: {
        print("서울특별시의 구:", $0)
    })
    .disposed(by: disposeBag)


----------merge2----------
서울특별시의 구: 강남구
서울특별시의 구: 강동구
서울특별시의 구: 영등포구
서울특별시의 구: 양천구
서울특별시의 구: 강북구
서울특별시의 구: 성북구
서울특별시의 구: 동대문구
서울특별시의 구: 종로구

 

zip

 

순서를 보장하면서 하나씩 합쳐진다.

마지막 "🇨🇳" 결과 값은 나오지 않았는데, 둘 중 하나의 Observable 이라도 완료되면 이 zip 전체가 완료된다. 

 

print("----------zip----------")
enum 승패 {
    case 승
    case 패
}

let 승부 = Observable<승패>.of(.승, .승, .패, .승, .패)
let 선수 = Observable<String>.of("🇰🇷", "🇨🇭", "🇺🇸", "🇧🇷", "🇯🇵", "🇨🇳")

let 시합결과 = Observable.zip(승부, 선수) { 결과, 대표선수 in
    return 대표선수 + " 선수 " + " \(결과)!"
}

시합결과
    .subscribe(onNext: {
        print($0)
    })
    .disposed(by: disposeBag)
    

----------zip----------
🇰🇷 선수  승!
🇨🇭 선수  승!
🇺🇸 선수  패!
🇧🇷 선수  승!
🇯🇵 선수  패!