[RxSwift] Operators (2)

 

 

Transforming Operators

 

toArray

 

Observable이 방출하는 모든 요소를 배열에 담은 다음 Completed 시점에 이 배열을 방출(하나의 요소를 방출)하고자 할때 toArray 연산자를 사용한다.

소스 Observable이 종료되기 전까지 구독자에게 이벤트가 전달되지 않는다.

 

let subject = PublishSubject<Int>()

subject
    .toArray()
    .subscribe {
        print($0)
    }
    .disposed(by: disposeBag)

subject.onNext(1) // 구독자로 전달되지 않음
subject.onNext(2) // 구독자로 전달되지 않음

subject.onCompleted() // Observable이 종료되어 구독자애게 하나의 배열을 전달하게 된다.

// success([1, 2])

 

map

 

Observable이 방출하는 항목을 대상으로 클로저를 실행하여 실행결과를 방출하는 Observable을 return 한다.

let colors = ["red", "blue", "yellow"]

Observable.from(colors)
    .map {
        "color is \($0)"
    }
    .subscribe {
        print($0)
    }
    .disposed(by: disposeBag)
.
.
.
// next(color is red)
// next(color is blue)
// next(color is yellow)
// completed

 

꼭 파라미터와 동일한 타입의 값을 return 해야하는건 아니다. (다른 타입으로 변형한뒤 전달 가능)

Observable.from(colors)
    .map {
        $0.count // String 타입이 아닌 Int 타입의 값도 return 할 수 있다.
    }
    .subscribe {
        print($0)
    }
    .disposed(by: disposeBag)
.
.
.
// next(3)
// next(4)
// next(6)
// completed

 

flatmap

 

[1]원본 Observable이 항목을 방출하면 flatmap 연산자가 변환 함수를 실행하고 변환함수는 방출된 항목을 Observable로 변환한다.

또한 [2]방출된 항목의 값이 추후에 바뀐다면 바뀐 값이 적용된채로 flatmap 연산자가 실행되어 다시또 새로운 항목을 방출한다.

 

즉 모든 Observable이 방출하는 항목을 모아서 최종적으로 하나의 Observable을 return 하면서도

-> Observable을 방출하는 Observable 스트림을 하나로 합치는 == swift에서 2차원 배열을 1차원적으로 flat화 하는 고차함수인 flatmap 과 맥락을 비슷하게 이해하면 좋을것같다. )

그 Observable들이 방출하는 항목을 지속적으로 감시하면서 최신값을 확인 할수 있다.

 

aSubject, bSubject, subject 이렇게 3개의 subject가 있다.

그리고 BehaviorSubject<Int>타입을 방출하는 PublishSubject인 subject를 flapmap을 거친후에 구독을 한 상태이다.

 

여기서 [1]원본 Observable인 subject가 새로운 항목( aSubject, bSubject )을 방출 했을때 

flatmap 연산자를 거친후에 방출된 항목을 새로운 Observable로 변환하게 된다.

후에 [2]방출된 항목의 값이 추후에 바뀐다면 다시 또 flatmap 연산자를 거처서 새로운 Observable이 생성된다.

let aSubject = BehaviorSubject(value: 1)
let bSubject = BehaviorSubject(value: 2)

let subject = PublishSubject<BehaviorSubject<Int>>() // BehaviorSubject를 방출하는 PublishSubject

subject
    .flatMap { 
        $0.asObservable() // [1] 원본 Observable이 방출하는 항목(BehaviorSubject<Int>)을 새로운 Observable로 변환
    }
    .subscribe {
        print($0)
    }
    .disposed(by: disposeBag)
.
.
.
// [1] PublishSubject에서 aSubject를 방출하면 flatmap 연산자를 거처서 새로운 Observable이 생성된다.
subject.onNext(aSubject)
// next(1)

// [1] PublishSubject에서 bSubject를 방출하면 flatmap 연산자를 거처서 새로운 Observable이 생성된다.
subject.onNext(bSubject)
// next(2)
.
.
.
// [2] 방출이력이 있었던 aSubject에 새로운 Next 이벤트가 방출 한다면 다시 또 flatmap 연산자를 거처서 새로운 Observable이 생성된다.
aSubject.onNext(10)
// next(10)

// [2] 방출이력이 있었던 bSubject에 새로운 Next 이벤트가 방출 한다면 다시 또 flatmap 연산자를 거처서 새로운 Observable이 생성된다.
bSubject.onNext(11)
// next(11)

 

scan

 

원본 Observable이 방출하는 항목을 대상으로 변환을 실행한 다음 결과를 방출하는 하나의 Observable을 return 한다.

첫번째 파라미터에는 기본값을 전달하고 두번째 파라미터에는 Accumulator 클로저를 전달 한다.

작업 결과를 누적시키면서 중간 결과와 최종결과가 모두 필요할때 사용한다.

최종적으로 계산되는 값만 필요할때는 reduce 연산자를 사용한다.

 

Observable.range(start: 1, count: 4) // 1에서 10까지 정수를 방출하는 Observable
    .scan(0, accumulator: +) // 기본값인 0부터 시작해서 연산을 시작한다.
    .subscribe {
        print($0)
    }
    .disposed(by: disposeBag)
.
.
.
// 0 + 1
// next(1)
// 1 + 2
// next(3)
// 3 + 3
// next(6)
// 6 + 4
// next(10)
// completed

 

 

Combining Operators

 

reduce

 

중간 결과까지 모두 방출하는 scan 연산자와 달리 reduce 연산자는 최종 결과 하나만 방출한다.

 

Observable.range(start: 1, count: 4)
    .reduce(0, accumulator: +)
    .subscribe {
        print($0)
    }
    .disposed(by: disposeBag)
.
.
.
// next(10)
// completed

 

merge

여러 Observable이 배출하는 항목들을 하나의 Observable에서 방출하도록 병합하고자 할때 merge 연산자를 사용한다.

2개이상의 Observable을 병합하고 모든 Observable에서 방출하는 요소들을 순서대로 방출하는 Observable을 return한다.

또한 해당 Observable 모두가 completed 이벤트가 전달되어야 Observable이 종료된다. (Error같은 경우 하나라도 발생하면 종료된다. )

 

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

Observable.of(oddNumbers, evenNumbers)
    .merge() // 2개 이상의 Observable이 방출하는 요소들을 병합한 하나의 Observable을 리턴한다.
    .subscribe {
        print($0)
    }
    .disposed(by: bag)

oddNumbers.onNext(3)
oddNumbers.onNext(4)
.
.
.
// next(1)
// next(2)
// next(3)
// next(4)

 

 

combineLatest

 

Oservable을 결합하고자 할때 combineLatest 연산자를 사용한다.

두개의 소스 Observable 모두 요소를 방출한 시점에 클로저를 실행하여 구독자에게 Next 이벤트를 전달하며 

그후에 두개의 소스 Observable 중에서 하나라도 요소를 방출한다면 계속적으로 가장 최근 요소를 대상으로 클로저를 실행한다

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)

greetings.onNext("Hi") // greetings에 이벤트를 전달하였지만 아직 languages에는 아무런 이벤트를 전달받지 못하여 구독자에게 전달되는 이벤트는 없다.

languages.onNext("World!") // languages에 next 이벤트를 전달하면 비로소 combineLatest 연산자 내의 클로저가 실행된다.

// next(Hi World!)

greetings.onNext("Hello") // greetings에 또 다른 이벤트를 전달하면 새로운 Oservable을 결합한 결과가 생성되고 구독자에게는 greetings의 최신 값인 "Hello"와 languages의 최신 값인 "World!"가 합친 상태로 이벤트가 전달된다.

// next(Hello World!)

greetings.onCompleted() // greetings에 onCompleted 이벤트가 전달되면 더이상 greetings에는 이벤트를 받지 않으며 가장 최신값인 "Hello"가 사용된다.

languages.onCompleted() // 모든 Observable이 onCompleted 이벤트를 전달 한다면 그때 비로소 구독자에게 onCompleted 이벤트를 전달한다.

// completed

// 하나라도 Error 이벤트를 전달하면 즉시 구독자에게 Error 이벤트를 전달하고 종료한다.

 

zip

 

소스 Observable 중에서 하나라도 요소를 방출하면 가장 최근 요소와 결합했던 combineLatest 연산자와는 달리

zip 연산자는 반드시 인덱스를 기준으로 짝을 일치시켜 전달한여 클로저에게 중복된 요소를 전달하지 않는다.

즉 첫번째 요소는 첫번째 요소와 결합, 두번째 요소는 두번째 요소와 결합을 시도한다.

 

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

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

numbers.onNext(1) // numbers에 이벤트를 전달하였지만 아직 strings에는 아무런 이벤트를 전달받지 못하여 구독자에게 전달되는 이벤트는 없다.

strings.onNext("one") // strings에 next 이벤트를 전달하면 비로소 zip 연산자 내의 클로저가 실행된다.
// next(1 - one)

numbers.onNext(2) // 2와 동일한 인덱스에 해당하는 짝이 없기 때문에 새로운 값을 방출했다 하더라도 구독자에게는 이전과 같은 값이 전달된다.
// next(1 - one)

strings.onNext("two") // 모든 소스 Observable이 2번째 요소를 방출하면 zip 연산자는 클로저를 실행하여 구독자에게 결과를 전달한다.
// next(2 - two)

numbers.onCompleted() // combineLatest와 달리 소스 Observable에서 하나라도 complete 이벤트를 전달하면 이후에는 각 인덱스에 해당하는 짝이 없기때문에 더이상 새로운 Next 이벤트를 전달하지 않게 된다.

strings.onCompleted() // 모든 Observable이 onCompleted 이벤트를 전달 한다면 그때 비로소 구독자에게 onCompleted 이벤트를 전달한다.

// completed

// 하나라도 Error 이벤트를 전달하면 즉시 구독자에게 Error 이벤트를 전달하고 종료한다.

 

 

Timer based Operators

 

interval

 

특정 주기마다 정수를 방출하는 Observable이 필요할때 interval 연산자를 사용한다.

첫번째 파라미터로 반복주기를 전달하고, 두번째 파라미터에는 정수를 방출할 scheduler 전달한다.

 

// 1초마다 정수를 방출하는 Observable을 생성
let intervalOservable = Observable<Int>.interval(.seconds(1), scheduler: MainScheduler.instance)

// 1초마다 정수를 방출하는 Observable을 구독
let interval = intervalOservable.subscribe {
    print($0)
}
.
.
.
// 1초마다 이벤트 발생
// next(0)
// next(1)
// next(2)
// next(3)
// next(4)
// next(5)...

// interval을 중단시키기 위해서는 직접 dispose 해야한다.
interval.dispose()

 

timer

 

첫번째 파라미터는 첫번째 요소가 방출되는 시점까지에 상대적인 시간이다.

즉 구독을 시작하고 첫번째 요소가 구독자에게 전달되는 시간을 나타낸다. (1초로 지정을 했다면 구독후 1초뒤에 요소가 전달된다.)

두번째 파라미터는 반복주기이며 기본값이 nil이다.

 이값에 따라서 타이머연산자의 동작방식이 달라진다. ( 기본값일때는 하나의 요소만 방출하고 종료한다. )

마지막 파라미터에는 timer가 동작할 scheduler를 전달한다

 

 

let timerOservable1 = Observable<Int>.timer(.seconds(1), scheduler: MainScheduler.instance)
    
timerOservable1
    .subscribe {
        print($0)
    }
    .disposed(by: bag)

// 반복주기는 기본값으로 설정된 nil이기 때문에 하나의 요소만 방출하고 종료한다.
// next(0)
// completed



let timerOservable2 = Observable<Int>.timer(.seconds(1), period: .seconds(1), scheduler: MainScheduler.instance)

let timer = timerOservable2
    .subscribe {
        print($0)
    }

// 두번째 파라미터에 1초라는 반복주기를 전달하였기 때문에 1초마다 1씩 증가하는 정수가 방출
// next(0)
// next(1)
// next(2)
// next(3)
// next(4)
// next(5)...

// interval을 중단시키기 위해서는 직접 dispose 해야한다.
timer.dispose()