RxSwift. Transforming Observable

Observable의 타입을 변경하거나 출력 결과를 다르게 하고 싶을 경우에 사용하는 방법에 대해서 알아보도록 하겠습니다.

1. buffer

Observable.of(1,2,3).buffer(timeSpan: 3, count: 2, scheduler: MainScheduler.instance)
    .subscribe(onNext: { value in
        print("value = \(value)")
    }).disposed(by: disposeBag)

말 그대로 이벤트를 버퍼에 저장한 뒤 묶어서 방출하는 방식입니다.

timeSpan : 버퍼에 저장되는 시간 간격

count : 버퍼에 저장되는 최대 이벤트 수

출력 결과는

value = [1, 2]
value = [3]

2. flatMap

Observable.of(1,2,3)
    .flatMap { data -> Observable<String> in
        let str = String(data)
        return Observable.just(str)
    }.subscribe(onNext: { text in
        print("text = \(text)")
    }).disposed(by: disposeBag)

map과 흡사한 동작을 합니다. 지금 들어온 이벤트를 가지고 다른 이벤트를 방출하는데 사용합니다.

그렇지만 다른점을 이야기하자면 map은 리턴되는 값이 Observable에 자동으로 넣어지게 되고, flatMap은 flat한 값들을 사용하게 됩니다.

3. flatMapFirst

Observable.of(1,2,3)
    .flatMapFirst { data -> Observable<String> in
        let str = String(data)
        return Observable.just(str)
    }.subscribe(onNext: { text in
        print("text = \(text)")
    }).disposed(by: disposeBag)

위의 flatMap과 흡사하게 동작하지만 새로 생성된 Observable이 첫번째 발생된 이후 끝날때까지 동작을 받지 않습니다.

무슨 의미냐면, 만약 네트워크 동작처럼 오래 걸리는 작업을 Observable에 걸었을 경우에 이 동작이 끝나기 전에 들어오는 다른 이벤트를 처리하지 않습니다.

그래서 저는 이걸로 소히 말하는 따닥을 막아버립니다.

4. flatMapLatest

Observable.of(1,2,3)
    .flatMapLatest { data -> Observable<String> in
        let str = String(data)
        return Observable.just(str)
    }.subscribe(onNext: { text in
        print("text = \(text)")
    }).disposed(by: disposeBag)

이것은 flatMapFirst와 거의 비슷한 동작을 하지만, 동작이 끝나기전에 다른 이벤트가 들어오게 된다면 이전 이벤트를 종료하여 버립니다.

dispose가 호출되지 않고 never처럼 동작하고 마지막 동작을 처리하도록 합니다.

5. map

Observable.of(1,2,3)
    .map { data -> String in
        return String(data)
    }.subscribe(onNext: { text in
        print("text = \(text)")
    })
    .disposed(by: disposeBag)

중간에 값을 변환하고 이것의 리턴형을 Observable 형태로 전달을 하게 됩니다.

6. scan

Observable.of(1,2,3)
    .scan(0, accumulator: { accumulator, num -> Int in
        return accumulator + num
    })
    .subscribe(onNext: { text in
        print("text = \(text)")
    })
    .disposed(by: disposeBag)

처음 seed를 지정하고 이 값들을 가지고 변형된 값을 만들수 있습니다.

여기서는 단순하게 1,2,3을 더한값을 방출하게 하였습니다.

또한 이것을 가지고 previous 값을 만들수 있는데요

func withPrevious(seed: E) -> Observable<(E, E)> {
    return scan((seed, seed)) { ($0.1, $1) }
}

이렇게 하면 튜플에 전달되는 앞에 값은 이전의 값이 되고, 튜플의 뒤에 값은 현재의 값이 됩니다.

7. window

Observable<Int>.range(start: 0, count: 10000)
    .window(timeSpan: 1000, count: 3, scheduler: MainScheduler.instance)
    .subscribe(onNext: { [weak self] observable in
        guard let `self` = self else { return }
        
        observable.subscribe(onNext: { number in
            print("text = \(number)")
        }, onDisposed: {
            print("disposed")
        }).disposed(by: self.disposeBag)
    })
    .disposed(by: disposeBag)

이것은 buffer와 거의 흡사합니다. 단지 다른점을 따지자면 나오는 값들이 observable로 나오게 됩니다.

...
 
text = 9117
text = 9118
text = 9119
disposed
text = 9120
text = 9121
text = 9122
disposed
 
...

결과는 이런형태가 되게 됩니다. 3개씩 잘라져 있죠?

8. reduce

Observable<Int>.range(start: 0, count: 10)
    .reduce(0, accumulator: { accumulator, number -> Int in
        return accumulator + number
    })
    .subscribe(onNext: { value in
        print("value = \(value)")
    })
    .disposed(by: disposeBag)

scan과 비슷한 동작을 하는데 다른점은 scan은 해당 결과가 나올때마다 방출하였지만 이것은 범위의 값이라면 다 처리가 되고 나서 한번만 방출합니다.

위의 예는 총합을 구하는 결과를 만들었는데요. 람다의 특성상 줄여서 표현도 가능합니다.

Observable<Int>.range(start: 0, count: 10)
    .reduce(0, accumulator: +)
    .subscribe(onNext: { value in
        print("value = \(value)")
    })
    .disposed(by: disposeBag)

즐 코딩~