[RxSwift] Operators - 변환(2)

Observable이 배출한 항목들을 변환하는 연산자들 - Buffer, FlatMap, Groupby, map, scan, window

 

안녕하세요 ◠‿◠ 고고입니다.

1. Buffer

• Buffer — Observable로부터 정기적으로 항목들을 수집하고 묶음으로 만든 후에 묶음 안에 있는 항목들을 한번에 하나씩 배출하지 않고 수집된 묶음 단위로 배출한다

구현부

public func buffer(timeSpan: RxTimeInterval, count: Int, scheduler: SchedulerType)
        -> Observable<[Element]> {
        BufferTimeCount(source: self.asObservable(), timeSpan: timeSpan, count: count, scheduler: scheduler)
    }

예시

let source = Observable<Int>.interval(RxTimeInterval.seconds(1), scheduler: MainScheduler.instance)

source.buffer(timeSpan: RxTimeInterval.seconds(3), count: 3, scheduler: MainScheduler.instance)
    .subscribe { print($0) }

// next([0, 1])
// next([2, 3, 4])
// next([5, 6, 7])
// ...

 

 

2. FlatMap

• FlatMap — 하나의 Observable이 발행하는 항목들을 여러개의 Observable로 변환하고, 항목들의 배출을 차례차례 줄 세워 하나의 Observable로 전달한다

구현부

public func flatMapLatest<Source: ObservableConvertibleType>(_ selector: @escaping (Element) throws -> Source)
        -> Observable<Source.Element> {
        return FlatMapLatest(source: self.asObservable(), selector: selector)
    }

예시

let source = Observable<Int>.interval(1.0, scheduler: MainScheduler.instance)
    
source.flatMap { num -> Observable<String> in
	return Observable<String>.create { observer in
		observer.on(.next("flatmap - \(num)"))
		return Disposables.create {
			print("dispose")
		}
	}
}.subscribe{ print($0) }
      
// next(flatpmap - 0)
// next(flatpmap - 1)
// next(flatpmap - 2)
// next(flatpmap - 3)
// ...

- FlatmapIndex == enumerate().flatMap

이벤트와 인덱스를 같이 받을수 있다.

 

- FlatMapFirst

원본 Observable의 이벤트를 받아 새로운 Observable 로 변형하지만, 이 Observable 이 완료되기 전에 이전 Observable 의 다른 이벤트들은 무시하게 된다.

 

- FlatMapLatest

원본 Observable의 이벤트를 받아 새로운 Observable 로 변형하지만, 다음 원본 이벤트가 발생하면 dispose 되고 다시 변형된 Observable 을 생성한다. subscription에 dispose 가 전달 되지는 않는다.

 

 

3. GroupBy

• GroupBy — 원본 Observable이 배출하는 항목들을 키(Key) 별로 묶은 후 Observable에 담는다. 이렇게 키 별로 만들어진 Observable들은 자기가 담고 있는 묶음의 항목들을 배출한다

구현부

public func groupBy<Key: Hashable>(keySelector: @escaping (Element) throws -> Key)
        -> Observable<GroupedObservable<Key, Element>> {
        GroupBy(source: self.asObservable(), selector: keySelector)
    }

예시

let source = Observable<Int>.interval(1, scheduler: MainScheduler.instance)

source.groupBy{ $0 % 2 == 0 }
	.flatMap { grouped -> Observable<String> in
		if grouped.key {
			return grouped.map{ "even \($0)" }
		} else {
			return grouped.map{ "odd \($0)" }
		}
	}.subscribe { print($0) }

// next(even 0)
// next(odd 1)
// next(even 2)
// next(odd 3)
// ...

 

4. Map

• Map — Observable이 배출한 항목에 함수를 적용한다

구현부

public func map<Result>(_ transform: @escaping (Element) throws -> Result)
        -> Observable<Result> {
        Map(source: self.asObservable(), transform: transform)
    }

예시

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

let source = Observable.from(numbers)

source.subscribe {
    print("number: \($0)")
}

// next(number: 1)
// next(number: 2)
// next(number: 3)
// next(number: 4)
// next(number: 5)
// completed

- mapWithIndex == enumerate().map 

이벤트와 인덱스를 같이 받을수 있다.

 

5. Scan

• Scan — Observable이 배출한 항목에 연속적으로 함수를 적용하고 실행한 후 성공적으로 실행된 함수의 리턴 값을 발행한다

구현부

public func scan<A>(into seed: A, accumulator: @escaping (inout A, Element) throws -> Void)
        -> Observable<A> {
        Scan(source: self.asObservable(), seed: seed, accumulator: accumulator)
    }

예시

let source = Observable<Int>.interval(1, scheduler: MainScheduler.instance)

source.scan(10) { accumulator, num -> Int in
	return accumulator + num
}.subscribe { print($0) }

// next(10)
// next(11)
// next(13)
// next(16)
// ...

 

6. Window

• Window — 정기적으로 Observable의 항목들을 더 작은 단위의 Observable 윈도우로 나눈 후에, 한번에 하나씩 항목들을 발행하는 대신 작게 나눠진 윈도우 단위로 항목들을 배출한다

구현부

public func window(timeSpan: RxTimeInterval, count: Int, scheduler: SchedulerType)
        -> Observable<Observable<Element>> {
            return WindowTimeCount(source: self.asObservable(), timeSpan: timeSpan, count: count, scheduler: scheduler)
    }

예시

let rangeObservable = Observable<Int>.range(start: 0, count: 10)

let windowTest = rangeObservable.window(timeSpan: 1000, count: 3, scheduler: MainScheduler.instance)

 windowTest.subscribe(onNext: { [unowned self] observable in
	observable.subscribe { event in
		print(event)
	}
})

// next(0)
// next(1)
// next(2)
// completed
// next(3)
// next(4)
// next(5)
// completed
// next(6)
// next(7)
// next(8)
// completed
// next(9)
// completed

 

 

+)

7. Reduce

•  Observable이 배출한 항목에 연속적으로 함수를 적용하고 실행한 후 성공적으로 실행된 함수의 최종결과를 발행한다
• 중간결과와 최종결과가 모두 필요하면 scan 연산자
• 최종결과만 필요하면 reduce 연산자

구현부

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

예시

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

print("== scan")

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

print("== reduce")

source.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

 

 

출처 : http://reactivex.io/documentation/ko/operators.html

ReactiveX - Operators