RxSwift(25). withLatestFrom, sample, switchLatest, reduce

withLatestFrom

이 연산자는 ✧ㅏ래 형태로 사용된다.

triggerObservable.withLatestFrom(dataObservable)

연산자를 호출하는 옵저버블을 트리거옵저버블이라부른다

파라미터로 전달하는 옵저버블을 데이터옵저버블이라부른다.

 

트리거옵저버블이 NE를 방출하면 데이터 옵저버블이 가장 최근에 방출한 NE를 구독자에게 전달한다.

예를들어 회원가입버튼을 탭하는 시점에 TextField에 입력된 값을 가져오는 기능을 구현할 때 활용할 수 있다.

public func withLatestFrom<Source: ObservableConvertibleType, ResultType>(_ second: 
	Source, resultSelector: @escaping (Element, Source.Element) 
    throws -> ResultType) -> Observable<ResultType> {
    WithLatestFrom(first: self.asObservable(), second: second.asObservable(), resultSelector: resultSelector)
    }

이 연산자는 두가지 형태로 사용된다.

1. 첫 번째 형태는 데이터옵저버블과 클로저를 파라미터로 받는다. 클로저에는 두 옵저버블이 방출하는 요소가 전달되고 결과가 리턴된다.

연산자가 최종적으로 리턴하는 옵저버블은 클로저가 리턴하는 결과를 방출한다.

 

<추가예정>

 

2. 두번째 형태는 트리거 옵저버블에서 NE를전달하면 파라미터로 전달된 데이터옵저버블에서 가장 최근에 방출한 NE를 가져온다.

이벤트에 포함된 요소를 방출하는 옵저버블을 리턴한다.

let trigger = PublishSubject<Void>()
let data = PublishSubject<String>()

trigger.withLatestFrom(data)
    .subscribe{ print($0) }
    .disposed(by: bag)

data.onNext("hello")

 

아직 trigger서브젝트에서 NE를 방출하지 않았기 때문에 데이터 서브젝트로 전달한 NE는 구독자에게 전달되지 않는다

trigger.withLatestFrom(data)
    .subscribe{ print($0) }
    .disposed(by: bag)

data.onNext("hello")
trigger.onNext(())

//next(hello)

트리거에게 NE를 전달하면 구독자에게 데이터에 전달한 NE가 구독자에게 전달된다.

트리거 서브젝트로 다시한번 NE를 전달하면 동일한이벤트가 구독자에게 전달된다.

trigger.withLatestFrom(data)
    .subscribe{ print($0) }
    .disposed(by: bag)

data.onNext("hello")
trigger.onNext(())
trigger.onNext(())

//next(hello)
//next(hello)

 

이 연산자는 트리거옵저버블로 NE가 전달되면 데이터옵저버블의 최신 NE를 구독자에게 전달한다.

지금처럼 NE를 반복적으로 전달하면 동일한 이벤트가 반복적으로 구독자에게 전달된다.

이미 전달된 이벤트라고해서 전달되지 않는건 아니다.

 

데이터서브젝트로 CE가 전달되면 구독자에게 CE가 전달되지 않는다.

trigger.withLatestFrom(data)
    .subscribe{ print($0) }
    .disposed(by: bag)

data.onNext("hello")
trigger.onNext(())
trigger.onNext(())

data.onCompleted()
trigger.onNext(())

//next(hello)
//next(hello)
//next(hello)

다시 트리거서브젝트로 NE를보내면 데이터 서브젝트로 전달된 마지막 이벤트는 CE다. 

하지만 이떄는 CE가아닌 NE가 구독자에게 전달된다.

반대로 여기에서 EE가 전달된다면 바로 구독자에게 EE가 전달된다.

 

이후에 트리거로 NE를 보내더라도 구독자에게 전달되지 않는다.

트리거에 CE에 전달한다면 데이터에 전달한것관 달리 바로 CE가 전달되고 종료된다.

EE도 마찬가지이다.

---

sample

이 연산자는 아래와 같은 형태로 사용된다.

dataObservable.sample(triggerObservable)

withLatestFrom연산자와 반대이다.

이건 데이터옵저버블에서 연산자를 호출하고 트리거 옵저버블을 파라미터로 전달한다.

 

트리거옵저버블에서 NE를 전달할때마다 데이터옵저버블에서 최신데이터를 전달한다. 이부분은 withLatestFrom연산자와 같다.

 

하지만 동일한 NE를 반복해서 방출하지 않는다는 특징이 있다.

let trigger = PublishSubject<Void>()
let data = PublishSubject<String>()

data.sample(trigger)
    .subscribe{ print($0) }
    .disposed(by: bag)

trigger.onNext(())

아직 데이터서브젝트로 전달된 NE가 없기 때문에 구독자로 전달되는 이벤트도 없다.

data.sample(trigger)
    .subscribe{ print($0) }
    .disposed(by: bag)

trigger.onNext(())
data.onNext("Hello")

데이터서브젝트로 NE를 보냈지만 바로 구독자에게 전달되는건 아니다.

data.sample(trigger)
    .subscribe{ print($0) }
    .disposed(by: bag)

trigger.onNext(())
data.onNext("Hello")

trigger.onNext(())
//next(Hello)

트리거 서브젝트로 NE를 전달한 경우에만 전달된다.

data.sample(trigger)
    .subscribe{ print($0) }
    .disposed(by: bag)

trigger.onNext(())
data.onNext("Hello")

trigger.onNext(())
trigger.onNext(())
//next(Hello)

트리거에 NE를 보내면 이번엔 아무것도 전달되지 않는다.

다시말해 NE가 방출되지 않는다.

 

sample연산자는 동일한 Event를 두번이상 방출하지 않는다.

 

데이터에 CE를 보내고 다시 트리거에 NE를 보내면 구독자에게 CE가 전달된다.

withLatestfrom은 최신 NE가 전달되었지만, sample연산자는 CE를 그대로 전달한다.

 

데이터옵저버블에서 EE를 방출하면 트리거에 NE가 방출되지 않더라도 즉시 구독자에게 전달된다.

---

switchLatest

연산자 이름처럼 가장최근 옵저버블이 방출하는 이벤트를 구독자에게 전달한다.

어떤 옵저버블이 가장 최근 옵저버블인지 이해하는게 핵심이다.

 

 public func switchLatest() -> Observable<Element.Element> {
        Switch(source: self.asObservable())
    }

이 연산자는 파라미터가 없다.

그리고 주로 옵저버블을 방출하는 옵저버블에서 사용한다.

소스옵저버블이 가장 최근에 방출한 옵저버블을 구독하고 여기에서 전달된 NE를 방출하는 새로운 옵저버블을 리턴한다.

 

 

문자열을 방출하는 두개의 서브젝트와 문자열을 방출하는 옵저버블을 방출하는 서브젝트다.

let a = PublishSubject<String>()
let b = PublishSubject<String>()

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

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

a.onNext("1")
b.onNext("2")

a, b, source에는 아무런 연관이 없다. 그래서 아무것도 출력되지 않는다.

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

a.onNext("1")
b.onNext("2")

source.onNext(a)

//next(RxSwift.PublishSubject<Swift.String>)

소스 서브젝트에 a서브젝트를 전달하게되면 source에서 옵저버블을 방출하게된다.

여기에 switchLatest를 추가해본다.

---

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

a.onNext("1")
b.onNext("2")

source.onNext(a)

그렇다면 소스에서 방출된 옵저버블이 ✧ㅏ직 없기 때문에 구독자로 전달되는것도 없다. 이렇게 되면 소스서브젝트로 전달된 최신 옵저버블이 a옵저버블이된다.

switchLatest연산자는 최신옵저버블인 a에서 전달하는 이벤트를 구독자에게 전달한다.

이전에 a로 전달했던 1은 출력되지 않지만 새로운 NE를 전달하면 즉시 구독자에게 전달된다.

b서브젝트는 소스의 최신옵저버블이 아니기 때문에 전달되지 않는다.

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

a.onNext("1")
b.onNext("2")

source.onNext(a)
a.onNext("2")
b.onNext("b")

//next(2)

 

b서브젝트를 최신옵저버블로 만들고 싶다면 소스에 b서브젝트를 NE로 전달하면 된다.

b서브젝트를 전달하면 switchLatest연산자는 a의 구독을 종료하고 b를 구독한다.

 

소스로 CE를 전달해야 구독자로 전달되고 종료된다.

최신옵저버블에 EE를 전달하면 즉시 전달되고 종료된다.

---

reduce

scan연산자와 비교해보면 쉽게 이해할 수 있다.

scan연산자는 기본값과 소스옵저버블이 방출하는 값을 대상으로 두번째 파라미터로 전달한 accmulator연산을 실행한다.

그다음 클로저의 실행결과를 옵저버블을 통해 방출하고 다시 클로저로 전달한다.

소스옵저버블이 새로운 요소를 방출하면 이전결과와 함께 클로저를 다시 실행한다.

이런 특징 때문에 소스옵저버블이 방출하는 이벤트의 수와 구독자로 전달되는 이벤트의 수가 같다.

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

 

public func reduce<A>(_ seed: A, accumulator: @escaping (A, Element) throws -> A)
        -> Observable<A> {
        Reduce(source: self.asObservable(), seed: seed, accumulator: accumulator, mapResult: { $0 })
    }

reduce연산자는 seed값과 accumulator 클로저를 파라미터를 받는다.

seed값과 소스옵저버블이 방출하는 값을 대상으로 클로저를 실행하고 결과옵저버블을 통해 결과를 방출한다.

reduce는 결과옵저버블을 통해 최종값 하나만 방출한다.

중간결과값까지 모두 방출하는 scan과는 차이가 있다.

 

3번째 파라미터를 받는 형식도 있다.

public func reduce<A, Result>(_ seed: A, accumulator: @escaping (A, Element) throws -> A, mapResult: @escaping (A) throws -> Result)
        -> Observable<Result> {
        Reduce(source: self.asObservable(), seed: seed, accumulator: accumulator, mapResult: mapResult)
    }

최종 결과를 다른 형식으로 바꾸고 싶을때 주로 사용된다.

reduce연산자 뒤에 map연산자를 연결하는 것과 동일한 패턴을 구현할 수 있다.

 

let o = Observable.range(start: 1, count: 5)

print("== scan")

o.scan(0, accumulator: +)
   .subscribe { print($0) }
   .disposed(by: bag)


print("== reduce")

o.reduce(0, accumulator: +)
    .subscribe{ print($0) }
    .disposed(by: bag)
    
/*
== scan
next(1)
next(3)
next(6)
next(10)
next(15)
completed
== reduce
next(15)
completed
*/