[RxSwift] Operator(1) - Filtering

이번 포스팅에서는 RxSwift의 Operator중에서 Filtering에 관련된 Operator들에 대해 알아보자. 

Filtering Operator를 통해 next 이벤트를 통해 받는 데이터를 선택적으로 취할 수 있다.

 

Ignoreing Operators

ignoreElements()

마블 다이어그램을 보면 알 수 있듯이, next를 통해 전달된 모든 데이터를 무시한다.

여기서 주의할 점은 next를 통해 들어온 이벤트만 무시하고,

completed, error 이벤트들은 정상적으로 전달되어 스트림이 끊기게 된다.

//Cold Observable의 경우
func observable() -> Observable<String> {
    return Observable.create { observer in
        observer.onNext("A")
        observer.onNext("B")
        observer.onNext("C")
        observer.onCompleted()
        
        return Disposables.create()
    }
}

//Filtering Operator는 next이벤트로 들어오는 데이터에 대한 처리이므로 Subscriber에서 수행됨.
observable()
    .ignoreElements()
    .subscribe(
        onNext: { print($0) },
        onCompleted: { print("completed") }
    )
    .disposed(by: disposeBag)
    
/* Prints:
 completed
*/

//Hot Observable (Subject)의 경우
let publish = PublishSubject<String>() //Input
let observable = publish.asObservable() //Output


observable
    .ignoreElements()
    .subscribe(
        onNext: { print($0) },
        onCompleted: { print("completed") }
    )
    .disposed(by: disposeBag)

publish.onNext("A")
publish.onNext("B")
publish.onNext("C")
publish.onCompleted()

/* Prints:
 completed
*/

 

elementAt

Observable에서 방출된 이벤트중 n번째 데이터만 취하고, Completed 된다.

 

observable
    .element(at: 1) //1번째 인덱스의 Element만 받겠다는 의미
    .subscribe(
        onNext: { print($0) },
        onCompleted: { print("completed") }
    )
    .disposed(by: disposeBag)

/* Prints:
 B
 completed
*/

 

 

filter

특정 조건을 만족하였을때, Subscriber에게 next이벤트의 데이터가 전달된다.

실제 filter의 구현을 살펴보면, 

public func filter(_ predicate: @escaping (Element) throws -> Bool)
    -> Observable<Element> {
    Filter(source: self.asObservable(), predicate: predicate)
}

predicate라는 어규먼트로 데이터가 필터링될 조건에 대한 클로저를 작성하면, 

해당 조건이 참일때만, 데이터가 방출이 된다.

observable
    .filter{ $0 == "B" }
    .subscribe(
        onNext: { print($0) },
        onCompleted: { print("completed") }
    )
    .disposed(by: disposeBag)

/* Prints:
 B
 completed
*/

 

 

Skipping Operators

skip

next이벤트로 전달받은 데이터 n개를 건너뛰어주는 연산자이다.

 

observable
    .skip(2) // 2개 (A, B)를 건너 뜀
    .subscribe(
        onNext: { print($0) },
        onCompleted: { print("completed") }
    )
    .disposed(by: disposeBag)

/* Prints:
 C
 completed
*/

 

skipWhile

특정 조건을 만족할 때까지, 방출되는 데이터를 skip한다.

skip 할 조건은 클로져를 통해 구현이 되는데, 

filter와 달리, 클로져의 조건이 false일 때 데이터의 방출을 시작한다.

observable
    .skipWhile { $0 != B } // B일때 false이므로, B부터 데이터 방출 시작
    .skip(while: { $0 != B })  //skipWhile이랑 같음.
    .subscribe(
        onNext: { print($0) },
        onCompleted: { print("completed") }
    )
    .disposed(by: disposeBag)

/* Prints:
 B
 C
 completed
*/

 

skipUntil

다른 Observable에서 next이벤트를 방출하기 전까진 방출하는 이벤트를 Skip 한다.

즉, 다른 Observable을 통해 데이터 방출의 시작 시점을 다이나믹하게 필터링할 수 있다.

let publish = PublishSubject<String>()
let observable = publish.asObservable()

let trigger = PublishSubject<String>()

observable
    .skipUntil(trigger) // trigger Observable에서 데이터가 방출되기 전까지 skip
    .skip(until: trigger) // skipUntil과 같음.
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)

publish.onNext("A")
publish.onNext("B")

trigger.onNext("!!!!")

publish.onNext("C")

/* Prints:
 C
*/

 

 

Taking Operators

Taking operator들을 Skip과 반대로 생각하면 쉽게 이해할 수 있다. 

take

take연산자는 방출된 요소에 대해서 n개의 데이터만 취한 후, completed(스트림이 끊기게)된다.

observable
    .take(1)
    .subscribe(
        onNext: { print($0) },
        onCompleted: { print("Completed") }
    )
    .disposed(by: disposeBag)
    
/* Prints:
 A
 completed
*/

 

takeWhile

조건이 false가 되기 전까지, 데이터를 취하게 된다. 

observable
    .takeWhile { $0 != "B" } // B 데이터에서 true가 되어, 해당 시점부터 방출을 멈춘다(Completed됨).
    .take(while: {$0 != "B" }) // takeWhile과 같음.
    .subscribe(
        onNext: { print($0) },
        onCompleted: { print("Completed") }
    )
    .disposed(by: disposeBag)

/* Prints:
 A
 completed
*/

 

takeUntil

다른 Observable(trigger)에서 데이터가 방출되기 전까지 take 하도록 하는 연산자이다.

observable
    .takeUntil(trigger)
    .take(until: trigger) // takeUntil과 같음.
    .subscribe(
        onNext: { print($0) },
        onCompleted: { print("Completed") }
    )
    .disposed(by: disposeBag)

publish.onNext("A")
publish.onNext("B")

trigger.onNext("!!!!")

publish.onNext("C")

/* Prints:
 A
 B
 completed
*/

RxCocoa와 takeUntil을 활용하여 dispose를 할 수 있다. 

observable
 	.takeUntil(self.rx.deallocated)  
       //self(view model or view controller)가 해제되기 전까지, 데이터를 취한다.
 	.subscribe(onNext: { print($0) })

 

takeLast

take는 앞에서부터 n개의 데이터를 취한 반면,

takeLast는 뒤에서부터 n개의 데이터를 취한다.

 

 

Distinct Operators

distinctUntilChanged

연속적으로 방출되는 값의 중복되는 데이터를 걸러주는 연산자이다. 

예를 들어 "1, 1, 2, 2, 1, 1, 3" 이 순차적으로 방출되었다고 한다면, 

해당 연산자를 통해 "1, 2, 1, 3" 만 값을 받아오게 된다.

let publish = PublishSubject<String>()
let observable = publish.asObservable()

observable
    .distinctUntilChanged()
    .subscribe(
        onNext: { print($0) },
        onCompleted: { print("Completed") }
    )
    .disposed(by: disposeBag)

publish.onNext("A")
publish.onNext("A")

publish.onNext("B")
publish.onNext("B")

publish.onNext("A")

publish.onNext("C")
publish.onNext("C")


/* Prints:
 A
 B
 A
 C
*/

또한, 해당 연산자는 keyPath를 활용하여 원하는 객체의 프로퍼티에 접근도 가능하며,

클로져를 통해, 비교 로직을 직접 구현할 수 있다. 

 

실제 distingUntilChanged의 definition을 살펴보면,

아래와 같이 오버로딩을 통해 상황에 맞게 사용이 가능하다. 

// 1
public func distinctUntilChanged()
    -> Observable<Element> {
    self.distinctUntilChanged({ $0 }, comparer: { ($0 == $1) })
}

// 2. 직접 객체 프로퍼티를 지정하여 비교
public func distinctUntilChanged<Key: Equatable>(_ keySelector: @escaping (Element) throws -> Key)
    -> Observable<Element> {
    self.distinctUntilChanged(keySelector, comparer: { $0 == $1 })
}

// 3. keyPath를 활용하여 비교할 프로퍼티 지정
public func distinctUntilChanged<Property: Equatable>(at keyPath: KeyPath<Element, Property>) ->
    Observable<Element> {
    self.distinctUntilChanged { $0[keyPath: keyPath] == $1[keyPath: keyPath] }
}

// 4. 클로져를 통한 커스텀 로직
public func distinctUntilChanged(_ comparer: @escaping (Element, Element) throws -> Bool)
    -> Observable<Element> {
    self.distinctUntilChanged({ $0 }, comparer: comparer)
}

// 5. 직접 객체 프로퍼티를 지정하여 비교로직 커스텀
public func distinctUntilChanged<K>(_ keySelector: @escaping (Element) throws -> K, comparer: @escaping (K, K) throws -> Bool)
    -> Observable<Element> {
        return DistinctUntilChanged(source: self.asObservable(), selector: keySelector, comparer: comparer)
}

 

class Person {
    let age: Int
    
    init(age: Int) {
        self.age = age
    }
}

아래와 같이 Person 타입의 객체를 방출하는 observable이 있다고 하자. 

func observable() -> Observable<Person> {
    return Observable.create { observer in
        observer.onNext(Person(age: 10))
        observer.onNext(Person(age: 10))
        
        observer.onNext(Person(age: 21))
        observer.onNext(Person(age: 21))
        
        observer.onNext(Person(age: 22))
        
        observer.onNext(Person(age: 10))
        
        observer.onNext(Person(age: 30))

        observer.onCompleted()
        
        return Disposables.create()
    }
}

 

이렇게 비교할 프로퍼티를 keyPath 혹은 클로져를 통해 직접 지정할 수도 있고, 

observable()
    .distinctUntilChanged {$0.age}   // // 2. 직접 객체 프로퍼티를 지정하여 비교
    .distinctUntilChanged(at: \.age) // 3. keyPath를 활용하여 비교할 프로퍼티 지정
    .subscribe(
        onNext: { print($0.age) },
        onCompleted: { print("Completed") }
    )
    .disposed(by: disposeBag)
    
/* Prints:
 10
 21
 22
 10
 30
 Completed
*/

 

직접 비교할 로직을 클로저를 통해 커스텀도 가능하다. 

observable()
    .distinctUntilChanged(comparer: { a, b in  // 4. 클로져를 통한 커스텀 로직
        
        if( (a.age / 10) == (b.age / 10)) {
            return true
        }
        else {
            return false
        }
    })
    .subscribe(
        onNext: { print($0.age) },
        onCompleted: { print("Completed") }
    )
    .disposed(by: disposeBag)
    
/* Prints:
 10
 21
 10
 30
 Completed
*/

 

아래와 같이 프로퍼티를 지정하는 클로져와 비교 로직을 커스텀하는 클로져를 통해 구현도 가능하다.

observable()
    .distinctUntilChanged({ $0.age }, comparer: { a, b in 
    // 5. 직접 객체 프로퍼티를 지정하여 비교로직 커스텀
        
        if( (a / 10) == (b / 10)) {
            return true
        }
        else {
            return false
        }
    })
    .subscribe(
        onNext: { print($0.age) },
        onCompleted: { print("Completed") }
    )
    .disposed(by: disposeBag)
    
/* Prints:
 10
 21
 10
 30
 Completed
*/

 

 

Reference

https://github.com/fimuxd/RxSwift/blob/master/Lectures/05_Filtering%20Operators/Ch5.%20FilteringOperators.md

https://www.notion.so/Wallaby-RxSwift-72194669a39a4557baa69c672268af38