7. Combining Operators

Prefixing and concatenating

• Observable로 작업할 때 가장 먼저 필요한 것은 observer가 초기 값을 받도록 보장하는 것
• "현재 상태"에 대한 정보가 필요한 상황이 있다. → 현재위치 또는 네트워크 연결상태

example(of: "startWith") {
    let numbers = Observable.of(2, 3, 4)            // numbers 시퀀스를 생성
 
    let observable = numbers.startWith(1)           // 시퀀스의 시작을 1로 설정하고 numbers가 연결
    _ = observable.subscribe(onNext: { value in
        print(value)
    })
}

기존의 numbers에 있는 (2, 3, 4) 앞에 1이 추가되어 출력

 

• 결과적으로 startWith(_: )는 보다 일반적인 concat 연산자 계열의 간단한 변형
• 초기값은 RxSwift가 startWith(_ : ) 연결되는 시퀀스를 추가하는 한 아이템의 시퀀스
• Observable.concat(_ : ) 정적함수는 두 시퀀스를 연결

example(of: "Observable.concat") {
    let first = Observable.of(1, 2, 3)
    let second = Observable.of(4, 5, 6)
 
    let observable = Observable.concat([first, second])
    observable.subscribe(onNext: { value in
        print(value)
    })
}

Observable.concat을 사용하면 다양한 시퀀스를 합칠 수 있다.

 

• 시퀀스를 추가하는 또 다른 방법은 concat(_: ) 연산자

example(of: "concat") {
    let germanCities = Observable.of("Berlin", "Münich", "Frankfurt")
    let spanishCities = Observable.of("Madrid", "Barcelona", "Valencia")
 
    let observable = germanCities.concat(spanishCities)
    _ = observable.subscribe(onNext: { value in
        print(value)
    })
}

germanCities 시퀀스에 이어서 spanishCities가 연결되어 출력

 

• concatMap(_: )은 클로저에 의해 생성된 각 시퀀스가 다음 시퀀스가 구독되기 전에 완료될 때까지 실행되도록 보장
• 따라서 concatMap(_: )은 순차 순서를 보장하는 편리한 방법

example(of: "concatMap") {
    let sequences = [
        "German cities": Observable.of("Berlin", "Münich", "Frankfurt"),
        "Spanish cities": Observable.of("Madrid", "Barcelona", "Valencia")
    ]
 
    let observable = Observable.of("German cities", "Spanish cities")
        .concatMap { country in sequences[country] ?? .empty() }
 
    _ = observable.subscribe(onNext: { string in
        print(string)
    })
}

---

Merging

• RxSwift는 시퀀스를 결합하는 여러 가지 방법을 제공
• 가장 쉬운 방법이 merge

example(of: "merge") {
    let left = PublishSubject<String>()
    let right = PublishSubject<String>()
 
    let source = Observable.of(left.asObservable(), right.asObservable())
 
    let observable = source.merge()
    let disposable = observable.subscribe(onNext: { value in
        print(value)
    })
 
    var leftValues = ["Berlin", "Munich", "Frankfurt"]
    var rightValues = ["Madrid", "Barcelona", "Valencia"]
 
    repeat {
        switch Bool.random() {
        case true where !leftValues.isEmpty:
            left.onNext("Left: " + leftValues.removeFirst())
        case false where !rightValues.isEmpty:
            right.onNext("Right: " + rightValues.removeFirst())
        default:
            break
        }
    } while !leftValues.isEmpty || !rightValues.isEmpty
 
    left.onCompleted()
    right.onCompleted()
}

• merge() Observable은 수신하는 각 시퀀스를 구독하고 도착하자마자 아이템을 방출
• 사전 정의된 순서는 없다.
  • merge() 는 source 시퀀스가 완료되고 모든 내부 시퀀스가 완료된 후에 완료
  • 내부 시퀀스가 완료되는 순서는 관련이 없다.
  • 시퀀스에서 오류가 발생하면 merge() Observable이 즉시 오류를 전달한 다음 종료

---

Combining Elements

• RxSwift의 필수 연산자 그룹은 combineLatest 와 관련된 연산자
• combineLatest는 여러 시퀀스의 값을 결합
• 결합된 내부 시퀀스 중 하나가 값을 내보낼 때마다 사용자가 제공하는 클로저를 호출
• 각 내부 시퀀스에서 보낸 마지막 값을 받는다.
• 예를 들어, 여러 텍스트필드를 한번에 관찰하고 해당 값을 결합하고 여러 소스의 상태를 확인하는 등 많은 구체적인 응용 프로그램이 있다.

example(of: "combineLatest") {
    let left = PublishSubject<String>()
    let right = PublishSubject<String>()
 
    let observable = Observable.combineLatest(left, right) { lastLeft, lastRight in
        "\(lastLeft) \(lastRight)"
    }
 
    let disposable = observable.subscribe(onNext: { value in
        print(value)
    })
 
    print("> Sending a value to Left")
    left.onNext("Hello,")
    print("> Sending a value to Right")
    right.onNext("world")
    print("> Sending another value to Right")
    right.onNext("RxSwift")
    print("> Sending another value to Left")
    left.onNext("Have a good day,")
 
    left.onCompleted()
    right.onCompleted()
}

• 각 시퀀스의 최신 값을 인수로 받는 클로저를 사용하여 Observable을 결합
• 결합된 각 관측 값이 하나의 값을 방출할 때까지 아무일도 일어나지 않는다. 그 후, 새로운 값을 방출할 때마다 클로저는 각 Observable의 최신 값을 수신하고 해당 아이템을 생성

 

• 일반적인 패턴은 값을 튜플에 결합한 다음 체인 아래로 전달하는 것
• 일반적으로 아래와 같이 값을 결합한 다음 filter(_: )를 호출하는 경우가 많다.

let observable = Observable
                    .combineLatest(left, right) { ($0, $1) }
                    .filter { !$0.0.isEmpty }

 

• combineLatest 계열의 연산자에는 여러 가지 변형이 있다.
• 2개에서 8개 사이의 Observable 시퀀스를 매개 변수로 사용

example(of: "combine user choice and value") {
    let choice: Observable<DateFormatter.Style> = Observable.of(.short, .long)
    let dates = Observable.of(Date())
 
    let observable = Observable.combineLatest(choice, dates) { formate, when -> String in
        let formatter = DateFormatter()
        formatter.dateStyle = format
        return formatter.string(from: when)
    }
 
    _ = observable.subscribe(onNext: { value in
        print(value)
    })
}

• 이 예제는 사용자 설정이 변경될 때마다 화면의 값이 자동 업데이트

 

• 또 다른 Combination Operator에는 zip 연산자 계열이 있다.
• combineLatest와 마찬가지로 여러 가지 변형이 있다.

example(of: "zip") {
    enum Weather {
        case cloudy
        case sunny
    }
 
    let left: Observable<Weather> = Observable.of(.sunny, .cloudy, .cloudy, .sunny)
    let right = Observable.of("Lisbon", "Copenhagen", "London", "Madrid", "Vienna")
 
    let observable = Observable.zip(left, right) { weather, city in
        return "It's \(weather) in \(city)"
    }
 
    _ = observable.subscribe(onNext: { value in
        print(value)
    })
}

• zip의 이점
  • 제공한 Observable을 구독
  • 각각이 새로운 값을 방출할 때까지 기다린다.
  • 두가지 새로운 값으로 클로저를 호출
• zip은 동일한 논리적 위치(1st와 1st, 2nd와 2nd 등)에서 각 Observable의 값을 쌍으로 만든다.
• 이는 논리적 위치가 맞지 않는 아이템이 있을 경우 zip은 더 이상 아무것도 보내지 않는다. → 따라서 위의 예에서 "Vienna"가 출력되지 않는다.
• 이를 indexed sequencing이라고 한다.
• 그러나 zip은 초기에 값 방출을 중지할 수 있지만 모든 내부 Observable이 완료될 때까지 자체적으로 완료되지 않아 각 항목이 작업을 완료할 수 있다.
• Swift에도 zip 연산자가 있지만 두 컬렉션의 항목에 대해서만 새로운 튜플 컬렉션을 만든다.
• RxSwift는 2~8개의 Observable에 대한 변형을 제공

---

Triggers

• 앱에는 다양한 요구 사항이 있으며 여러 입력 소스를 관리해야 한다.
• 한 번에 여러 Observable의 입력을 받아야 하는 경우가 많다.
• 일부는 단순히 코드에서 작업을 트리거하고 다른 일부는 데이터를 제공

 

• withLatestFrom(_: )

example(of: "withLatestFrom") {
    let button = PublishSubject<Void>()
    let textField = PublishSubject<String>()
 
    let observable = button.withLatestFrom(textField)       // 버튼이 값을 내보내면 무시하지만 대신 텍스트필드에서 받은 최신 값을 내보낸다.
    _ = observable.subscribe(onNext: { value in
        print(value)
    })
 
    textField.onNext("Par")
    textField.onNext("Pari")
    textField.onNext("Paris")
    button.onNext(())
    button.onNext(())
}

 

 

• sample(_ : ) → withLatestFrom(_: ) 과 유사
• 다른 Observable에서 최신 값을 방출하지만 마지막 "tick" 이후에 도착한 경우에만 방출

 

example(of: "sample") {
    let button = PublishSubject<Void>()
    let textField = PublishSubject<String>()
 
    let observable = textField.sample(button)           // textField 이벤트를 받다가 button 이벤트가 발생했을 때 값 방출
    _ = observable.subscribe(onNext: { value in
        print(value)
    })
 
    textField.onNext("Par")
    textField.onNext("Pari")
    textField.onNext("Paris")
    button.onNext(())
    button.onNext(())
}

 

• withLatestFrom(_: ) 은 Observable 데이터를 매개 변수로 사용하고 sample(_: )은 Observable 트리거를 매개 변수로 사용
• 트리거를 기다리는 것은 UI 작업을 할 때 큰 도움

---

Switches

• RxSwift에는 두 가지 주요 Switching Operator인 amb(_: ) 및 switchLatest() 가 있다.
• 둘 다 결합된 또는 source 시퀀스의 이벤트 간에 전환하여 Observable 시퀀스를 생성할 수 있다.
• 이를 통해 구독자가 런타임에 수신할 시퀀스의 이벤트를 결정할 수 있다.

 

• amb(_: ) → amb를 ambiguous로 생각

example(of: "amb") {
    let left = PublishSubject<String>()
    let right = PublishSubject<String>()
 
    let observable = left.amb(right)
    let _ = observable.subscribe(onNext: { value in
        print(value)
    })
 
    left.onNext("Lisbon")
    right.onNext("Copenhagen")
    left.onNext("London")
    left.onNext("Madrid")
    right.onNext("Vienna")
 
    left.onCompleted()
    right.onCompleted()
}

• amb(_: ) 연산자는 left 및 right Observable 항목을 구독
• 그둘 중 하나가 아이템을 방출할 때까지 기다린 다음 다른 아이템에서 구독을 취소
• 그 후에는 첫 번째 활성 Observable의 아이템만 전달
• 중복 서버에 연결하고 먼저 응답하는 서버를 고수하는 것과 같은 몇 가지 실용적인 응용 프로그램을 선택할 수 있다.

 

• switchLatest() → 더 많은 곳에서 사용되는 옵션

example(of: "switchLatest") {
    let one = PublishSubject<String>()
    let two = PublishSubject<String>()
    let three = PublishSubject<String>()
 
    let source = PublishSubject<Observable<String>>()
 
    let observable = source.switchLatest()
    let disposable = observable.subscribe(onNext: { value in
        print(value)
    })
 
    source.onNext(one)                                              // one을 선택하면서 one에 대한 이벤트를 출력
    one.onNext("Some text from sequence one")
    two.onNext("Some text from sequence two")
 
    source.onNext(two)                                              // two를 선택하면서 two에 대한 이벤트를 출력
    two.onNext("More text from sequence two")
    one.onNext("and also from sequence one")
 
    source.onNext(three)                                            // three를 선택하면서 three에 대한 이벤트를 출력
    two.onNext("Why don't you see me?")
    one.onNext("I'm alone, help me")
    three.onNext("Hey, it's three. I win.")
 
    source.onNext(one)                                              // 다시 one을 선택하면서 one에 대한 이벤트를 출력
    one.onNext("Nope. It's me, one!")
 
    disposable.dispose()
}

 

• 기존에 배운 flatMapLatest와 거의 똑같은 일을 한다.
• flatMapLatest는 최신 값을 Observable에 매핑한 다음 구독
• switchLatest와 마찬가지로 최신 구독만 활성 상태를 유지

---

Combining elements within a sequence

• reduce(_ : _ : ) → Swift 표준 라이브러리에도 구현되어 있다.

 

• Swift의 컬렉션과 비슷하지만 Observable 시퀀스를 사용

example(of: "reduce") {
    let source = Observable.of(1, 3, 5, 7, 9)
 
    let observable = source.reduce(0, accumulator: +)
    _ = observable.subscribe(onNext: { value in
        print(value)
    })
 
    // 동일한 코드
    print("===============================")
 
    let observable = source.reduce(0) { summary, newValue in
        return summary + newValue
    }
    _ = observable.subscribe(onNext: { value in
        print(value)
    })
}

• 제공한 초기값(여기서는 0)으로 시작
• Observable이 항목을 내보낼 때마다 reduce(_: _: )는 클로저를 호출하여 새로운 summary를 생성
• source Observable이 완료되면 reduce(_: _: )가 summary 값을 방출하고 완료

Note
reduce(_: _: )는 source Observable이 완료될 때만 누적 값을 생성.완료되지 않은 시퀀스에 이 연산자를 적용하면 아무것도 내보내지 않는다.

 

• scan(_: _: ) → reduce(_: _: )와 비슷하게 동작하나 값을 바로 방출

example(of: "scan") {
    let source = Observable.of(1, 3, 5, 7, 9)
 
    let observable = source.scan(0, accumulator: +)
    _ = observable.subscribe(onNext: { value in
        print(value)
    })
}

입력 값당 하나의 출력 값을 얻는다.

• source Observable이 아이템을 방출할 때마다 scan(_: accumulator: )은 클로저를 호출
• 새 아이템과 함께 실행 값을 전달하고 클로저는 새로운 누적 값을 반환
• scan 사용사례는 상당히 크다.
• scan Observable 상태 정보를 캡슐화하는 것은 좋은 아이디어. → 로컬 변수를 사용할 필요가 없으며 source Observable이 완료되면 사라진다.