[RxSwift] Combining Operators

Combining Operators 에 관해 정리한 내용을 기록합니다.

 

Combining Operators 

 

startWith

요소들의 스트림을 방출하기 전에 설정한 요소를 처음으로 시작하여 스트림을 진행할 수 있다. 

현재 상태, 위치나 네트워크 연결 상태 등과 같은 데에 사용한다.

 

Observable.of(2, 3)
    .startWith(1)
    .subscribe(onNext: {
        print($0)
    })
    .disposed(by: bag)

/*
1
2
3
*/

 

concat

두 개의 스트림을 연결한 것이다.

 

let firstObservables = Observable.of(1,1,1)
let secondObservables = Observable.of(2,2)

firstObservables
    .concat(secondObservables)
    .subscribe(onNext: {
        print($0)
    })
    .disposed(by: bag)

/*
1
1
1
2
2
*/

 

concatMap 

각각의 스트림이 다음 스트림이 구독되기 전에 합쳐지는 것을 보장한다. (순서가 보장된다.)

flatMap과 concat이 합쳐진 것과 같다. 

 

let sequences = ["Number": Observable.of("1", "2", "3"),
                 "Alphabet": Observable.of("A", "B", "C")]

Observable.of("Number", "Alphabet")
    .concatMap({
        sequences[$0] ?? .empty()
    })
    .subscribe(onNext: {
        print($0)
    })
    .disposed(by: bag)

/*
1
2
3
A
B
C
*/

 

merge

여러 개의 스트림을 합치는데 순서를 보장하지 않는다. 

merge(maxConcurrent:): 합치는 스트림의 개수를 제한하기 위해서 사용하고 네트워크 요청이 많아질 때 리소스를 제한하거나 연결 수를 제한한다.

 

let odds = Observable.of(1, 3, 5, 7)
let evens = Observable.of(2, 4, 6)

Observable.of(odds, evens)
		.merge()
    .subscribe(onNext: {
        print($0)
    })
    .disposed(by: bag)

/*
1
3
2
5
4
7
6
*/

Observable.of(odds, evens)
    .merge(maxConcurrent: 1)
    .subscribe(onNext: {
        print($0)
    })
    .disposed(by: bag)

/*
1
3
5
7
2
4
6
/*

 

combineLatest

결합된 스트림은 각각의 서브 스트림에서 이벤트가 발생할 때마다 새로 결합된 요소를 방출한다.

 

let left = PublishSubject<Int>()
let right = PublishSubject<String>()

Observable.combineLatest(left, right) {
    "\($0)+\($1)"
}
.subscribe(onNext: {
    print($0)
})
.disposed(by: bag)

left.onNext(1)
right.onNext("A")
left.onNext(2)
right.onNext("B")
right.onNext("C")
right.onNext("D")
left.onNext(3)
left.onNext(4)

/*
1+A
2+A
2+B
2+C
2+D
3+D
4+D
*/

let dateDisplayFormat = Observable<DateFormatter.Style>.of(.short, .long)
let currentDate = Observable<Date>.of(Date())

Observable.combineLatest(
    dateDisplayFormat,
    currentDate,
    resultSelector: { format, date -> String in
        let dateFormatter = DateFormatter()
        dateFormatter.dateStyle = format
        return dateFormatter.string(from: date)
    }
)
.subscribe(onNext: {
    print($0)
})
.disposed(by: bag)

/*
2022/01/02
January 2, 2022
*/

 

zip

결합된 스트림은 각각의 내부 스트림들이 모두 새로운 이벤트를 방출할 때까지 기다린 후 새로 결합하여 요소를 방출한다.

 

let left = Observable.of(1, 2, 3, 4)
let right = Observable.of("A", "B", "C", "D")

Observable.zip(left, right) {
    "\($0)+\($1)"
}
.subscribe(onNext: {
    print($0)
})
.disposed(by: bag)

/*
1+A
2+B
3+C
4+D
*/

 

withLatestFrom

Trigger 역할을 하는 Observable이 요소를 방출할 때 또 다른 하나의 Observable의 가장 최신 요소를 결합하여 방출한다.

 

let button = PublishSubject<Void>()
let textdField = PublishSubject<String>()

let observable = button.withLatestFrom(textdField)
    .subscribe(onNext: {
        print($0)
    })
    .disposed(by: bag)

textdField.onNext("H")
textdField.onNext("Her")
textdField.onNext("Here")
button.onNext(())
button.onNext(())

/*
Here
Here
*/

 

sample

withLatestFrom과 동작이 비슷하지만, Trigger 역할을 하는 Observable이 여러 번 요소를 방출해도 한 번만 방출한다. 

 

let button2 = PublishSubject<Void>()
let textField2 = PublishSubject<String>()

textField2.sample(button2)
    .subscribe(onNext: {
        print($0)
    })
    .disposed(by: bag)

textField2.onNext("I")
textField2.onNext("I am")
button2.onNext(())
button2.onNext(())

 

amb

ambiguous(모호한)이란 뜻을 가지고 있다.

두 개의 스트림 중 먼저 요소를 방출하는 스트림만 방출되게 한다. (다른 스트림의 요소 방출은 무시한다.)

 

let firstSubject = PublishSubject<Int>()
let secondSubject = PublishSubject<Int>()

firstSubject.amb(secondSubject)
    .subscribe(onNext: {
        print($0)
    })
    .disposed(by: bag)

firstSubject.onNext(1)
secondSubject.onNext(20)
firstSubject.onNext(2)
secondSubject.onNext(40)
firstSubject.onNext(3)
secondSubject.onNext(60)

/*
1
2
3
/*

 

switchLatest

여러 스트림 중 가장 최신의 이벤트가 발생된 Observable로 변경하여 방출한다. 

 

let one = PublishSubject<String>()
let two = PublishSubject<String>()
let source = PublishSubject<Observable<String>>()

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

source.onNext(one)
one.onNext("This is one.")
one.onNext("is This one?")
source.onNext(two)
two.onNext("This is two.")
two.onNext("is This two?")
source.onNext(one)
one.onNext("This is one.")

 

reduce

Swift에서의 reduce와 동일하게 작동한다. 

 

Observable.of(1, 2, 3, 4, 5)
    .reduce(0, accumulator: +)
    .subscribe(onNext: {
        print($0)
    })
    .disposed(by: bag)

// 15

 

scan

이전에 방출된 요소와 새로 방출된 요소를 결합하여 적용된 로직을 계속 누적한다.

매번 값이 들어올 때마다 로직이 적용된 결과 값을 방출한다.

총합, 통계, 상태를 계산할 때 등 다양하게 사용한다.

 

let nums = Observable.of(1, 2, 3, 4, 5)
let scan = nums.scan(0, accumulator: +)

Observable.zip(nums, scan) {
     "nums: \($0), scan: \($1)"
}
.subscribe(onNext: {
    print($0)
})
.disposed(by: bag)

/*
nums: 1, scan: 1
nums: 2, scan: 3
nums: 3, scan: 6
nums: 4, scan: 10
nums: 5, scan: 15
*/

 

 

부족한 점 피드백해주시면 감사합니다👍