RxSwift(24). startWith, concat, merge, combineLatest, zip

startWith

startWith는 옵저버블이 요소를 방출하기 전에 다른 항목들을 앞부분에 추가한다 주로 기본값이나 시작값을 지정할 때 활용된다.

public func startWith(_ elements: Element ...)
        -> Observable<Element> {
            return StartWith(source: self.asObservable(), elements: elements)
    }

startWith연산자의 파라미터는 가변파라미터이다. 파라미터로 전달하는 하나 이상의 값을 옵저버블 시퀀스 앞부분에 추가한다.

그런다음 새로운 옵저버블을 리턴한다.

let numbers = [1, 2, 3, 4, 5]

Observable.from(numbers)
    .startWith(-1, -2)
    .startWith(-3)
    .subscribe{ print($0) }
    .disposed(by: bag)
    
/*
next(-3)
next(-1)
next(-2)
next(1)
next(2)
next(3)
next(4)
next(5)
completed

*/

startwith는 연산자이고 두개 이상의 연산자를 이어 사용할 수 있다.

 

startwith는 기존 옵저버블 앞에 값을 추가한다.

연달아 사용하게 되면 추가된 새로운 옵저버블 앞에 또 새로운 값을 추가하게 된다.

 

last in first out이다.

---

concat

이 연산자는 두개의 옵저버블을 연결할 때 사용된다.

concat연산자는 타입메서드와 인스턴스메서드로 구현되어 있다.

 

타입메서드

타입메서드로 구현된 concat연산자는 파라미터로 전달된 컬렉션에 있는 모든 옵저버블을순서대로 연결한 하나의 옵저버블로 리턴한다.

let fruits = Observable.from(["🍏", "🍎", "🥝", "🍑", "🍋", "🍉"])
let animals = Observable.from(["🐶", "🐱", "🐹", "🐼", "🐯", "🐵"])

Observable.concat([fruits, animals])
    .subscribe{ print($0) }
    .disposed(by: bag)
/*
next(🍏)
next(🍎)
next(🥝)
next(🍑)
next(🍋)
next(🍉)
next(🐶)
next(🐱)
next(🐹)
next(🐼)
next(🐯)
next(🐵)
completed
*/

실행된 결과를 보면 과일이 모두 방출된 뒤에 동물이 방출된다. 또한 completedevent는 연결된 모든 요소를 방출한 후에 전달된다.

 

인스턴스메서드

이 연산자는 대상 옵저버블이 completedevent를 전달한 다음에 파라미터로 전달한 옵저버블을 연결한다.

만약 에러이벤트가 전달된다면 연결되지 않는다. 대상 옵저버블이 방출하는 요소만 전달되고 에러이벤트가 전달된 다음에 바로 종료된다.

이건 타입메서드로 구현된 concat연산자도 마찬가지이다.

 

fruits.concat(animals)
    .subscribe{print($0)}
    .disposed(by: bag)
    
/*
next(🍏)
next(🍎)
next(🥝)
next(🍑)
next(🍋)
next(🍉)
next(🐶)
next(🐱)
next(🐹)
next(🐼)
next(🐯)
next(🐵)
completed
*/

---

merge

여러 옵저버블이 배출하는 항목들을 하나의 옵저버블에서 방출하도록 병합한다. 

이 연산자는 concat연산자와 자주 혼동하게 되는데, concat과 동작방식이 다르다.

두개이상의 옵저버블을 병합하고 모든 옵저버블에서 방출하는 요소들을 순서대로 방출하는 옵저버블을 리턴한다.

 

이 연산자는 두개이상의 옵저버블이 방출하는 요소를 병합한 하나의 옵저버블을 리턴한다.

단순히 뒤에 연결하는 것이 아니라 하나의 옵저버블로 합쳐준다.

그래서 옵저버블이나 서브젝트로 전달된 이벤트가 순서대로 구독자에게 전달된다.

 

let oddNumbers = BehaviorSubject(value: 1)
let evenNumbers = BehaviorSubject(value: 2)
let negativeNumbers = BehaviorSubject(value: -1)

let source = Observable.of(oddNumbers, evenNumbers)
source.subscribe{ print($0) }
.disposed(by: bag)

//next(RxSwift.BehaviorSubject<Swift.Int>)
//next(RxSwift.BehaviorSubject<Swift.Int>)
//completed

merge

let source = Observable.of(oddNumbers, evenNumbers)

source
    .merge()
    .subscribe{ print($0) }
    .disposed(by: bag)
    
//next(1)
//next(2)

서브젝트가 아닌 요소를 방출한다.

 

이 상태에서 새로운 nextevent를 전달하게 되면 

let source = Observable.of(oddNumbers, evenNumbers)

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

oddNumbers.onNext(3)
evenNumbers.onNext(4)

//next(1)
//next(2)
//next(3)
//next(4)

3과 4가 순서대로 전달된다.

 

순서가 바뀐다면 출력 순서도 바뀐다.

 

oddNumbers에서 completedevent를 보내면 더이상 oddNumber서브젝트는 종료된다.

하지만 evenNumbers는 살아있기 때문에 즉시 구독자에게 전달된다.

 

merge로 만들어진 옵저버블은 포함된 모든 옵저버블이 종료(completed or errer)가 되면 최종적으로 종료된다.

그전까지는 계속 nextevent가 전달된다. 

 

병합대상중에서 하나라도 에러이벤트가 전달되면 그즉시 에러이벤트가 전달되고 종료된다.

 

merge연산자로 병합할 수 있는 옵저버블의 수는 제한되어있지 않다.

만약 제한해야 한다면

source
	.merge(maxConcurrent: 2)

를 사용하면 2개로 제한이 된다.

 

merge연산자는 큐에 병합대상들을 저장해놓는다. 순서대로 병합리스트에 넣게되는데

observable.of(1,2,3)

.merge(maxConcurrent: 2)

 

라면 1, 2를 큐에 넣고 1 or 2 가 completed되면  대상을 병합대상에서 제외하고 다음대기순서인  3을 추가하게된다

---

combineLatest

 

1에서 5까지 방출하는 옵저버블과 ABCD문자를 방출하는 옵저버블이있다.

이 두 옵저버블을 소스옵저버블이라고 한다.

 

combineLatest연산자는 소스옵저버블을 combine(결합)한다.

 

소스옵저버블을 결합한다음 파라미터로 전달된 함수를 실행하고 결과를 방출하는 새로운 옵저버블을 리턴한다.

다이어그램에선 숫자와 문자를 연결해서 문자를 리턴하고있다.

 

combinelatest의 핵심은 연산자가 리턴한 옵저버블이 언제 이벤트를 방출하는지이다.

아래에서 옵저버블이 연산자가 리턴한 옵저버블인데 이옵저버블을 결과옵저버블이라고 부른다.

 

1이 방출된 시점에 문자는 방출되지 않았기 때문에 결과옵저버블은 방출하지 않는다.

이어서 문자옵저버블이 A를 방출하면 두 옵저버블이 이벤트를 하나씩 방출한것이다. 바로 이 시점에 두 소스 옵저버블이 방출한 최신 이벤트를 대상으로 이너옵저버블의 함수를 실행한다.  여기에서는 숫자와 문자를 연결하게되고 이 결과가 연산자가 리턴한 옵저버블을 통해 구독자에게 전달된다. 그래서 구독자는 1A가 저장된 이벤트를 받는다.

이후에는 소스옵저버블에서 하나라도 이벤트를 방출하면 결과 옵저버블 역시 이벤트를 방출한다.

 

public static func combineLatest<O1: ObservableType, O2: ObservableType>
        (_ source1: O1, _ source2: O2, resultSelector: @escaping (O1.Element, O2.Element) throws -> Element)
            -> Observable<Element> {
        return CombineLatest2(
            source1: source1.asObservable(), source2: source2.asObservable(),
            resultSelector: resultSelector
        )
    }

combineLatest는 두개의 옵저버블과 클로저를 파라미터로 받는다. 옵저버블이 NE를 통해 전달하는 요소들은 클로저파라미터를 통해 파라미터로 전달된다. 그리고 이 클로저는 실행결과를 리턴하고 연산자는 최종적으로 이 결과를 방출하는 옵저버블을 리턴한다.

다양한 오버로딩도 존재한다.

 

클로저를 전달하지 않으면 옵저버블이 방출하는 요소들을 하나의 튜플로 합친다음 이 튜플을 방출하는 옵저버블을 리턴한다.

옵저버블을 최대 6개까지 전달할 수 있는 연산자들이 선언되어있다.

파라미터의 수만 다르고 동작방식은 앞과 동일하다.

let greetings = PublishSubject<String>()
let languages = PublishSubject<String>()

Observable.combineLatest(greetings, languages){ lhs, rhs -> String in
    return "\(lhs) \(rhs)"
}
.subscribe{ print($0) }
.disposed(by: bag)

해당 combineLatest는 두개의 PublishSubject로 구성되어있다.

아직 아무 이벤트도 전달되지 않았기 때문에 구독자에게 전달되지 않는다.

greetings과 languages가 둘 다 이벤트를 전닯받게 된다면 그 때부터 시작이다.

구독즉시 전달받고싶다면 startWith로 초기값을 지정해주거나, BehaviorSubject로 변경하면 된다.

 

둘중 하나에게만 Completedevent가 전달된다면 종료되지 않고 해당 서브젝트만 종료되고 나머지에게 NE를 전달시 종료되기전 값을 유지한채 이어진다.

 

둘 모두 CE가 전달되면 이 시점에 구독자에게 CE가 전달된다.

 

만약 둘 중 하나라도 EE가 전달되면 그 즉시 구독자에게 EE를 전달하고 종료된다.

---

zip

zip연산자는 combinelatest와 비교하면 된다.

정수옵저버블과 문자옵저버블이 있다.

소스옵저버블이 방출하는 요소를 결합한다.

옵저버블을 결합하고 클로저를 실행하고 이 결과를 방출하는 결과옵저버블을 리턴한다.

zip연산자는 클로저에게 중복된 값을 전달하지 않는다.

반드시 인덱스를 기준으로 짝을 일치시켜서 전달한다.

첫 번쨰는 첫번째끼리 결합, 두 번쨰는 두 번째끼리 결합.

만약 결합할 짝이 없다면 구독자에게 전달되지 않는다.

 

이처럼 소스옵저버블이 방출하는 요소를 순서를 일치시켜서 결합하는것을 Indexed Sequencing이라고 한다.

let numbers = PublishSubject<Int>()
let strings = PublishSubject<String>()

Observable.zip(numbers, strings){ "\($0) - \($1)" }
.subscribe{print($0)}
.disposed(by: bag)

numbers.onNext(1)
strings.onNext("one")

numbers.onNext(2)

//next(1 - one)

zip은 항상 순서로 짝을 맞추기 때문에 2는 같이 짝이 맞는 이벤트가 들어올 때 까지 순서를 기다린다.

combinelatest와 달리 하나라도 CE가 들어오면 이후에는 아무런 NE가 전달되지 않는다.

 

구독자에게 CE가 전달되는 시점은 모든 소스옵저버블에 CE가 전달되는 시점이다.

소스옵저버블중에서 하나라도 EE가 전달되면 즉시 구독자에게 EE가 전달되고 종료된다.