RxSwift(3)-Filtering Operators & TransForming Operators

next 이벤트를 통해 넘어온 값들에 대해 필터를 해줘 해당 오퍼레이터를 통해 넘어온 값들에 대해 다 처리하지 않고 원하는 값만 골라 처리할 수 있어지는 기능에 대해 살펴보겟습니다

이것도 4가지 종류가 있습니다.. 자주 나오는것 같다 4.4..

1. Ignoring Operator

ignoreElement()

값들이 방출되어도 무시되고 스트림에 등록되지 않는다. 하지만 error나 completed는 무시하지 않습니다.

example(of: "ignoreElements") {
  // 1
  let strikes = PublishSubject<String>()
  let disposeBag = DisposeBag()
// 2
  strikes
    .ignoreElements()
    .subscribe { _ in
      print("You're out!")
    }
    .disposed(by: disposeBag)
}
strikes.onNext("X")
strikes.onNext("X")
strikes.onCompleted()

--- Example of: ignoreElements ---
You're out!
결과는 마무리에만 찍힌다.

.elementAt(index)

원하는 index만 필터링 통과

Observable.of("A", "B", "C", "D")
    .elementAt(2)
    .subscribe(onNext: { value in
        print(value)
    })
    .disposed(by: disposeBag)
// 출력: C

-> 특정 순서의 이벤트나 데이터만 처리할때 사용할수 있을거같다

.filter

조건에 해당하는 값만 방출

2. Skip 연산자

Skip 연산자는 특정 조건이나 개수에 해당하는 이벤트를 건너뛰고, 이후의 이벤트만 처리합니다. 이는 filter와 중요한 차이점이 있습니다:

• filter: 각 요소마다 조건을 평가하여 통과 여부 결정
• skip: 특정 시점이나 조건까지 건너뛰고, 이후부터는 모든 요소 통과

.skip(_ : Int)

n개 만큼의 요소가 skip

// 첫 3초 간의 위치 업데이트는 무시 (GPS가 안정화되는 시간)
locationManager.rx.location
    .skip(3)
    .subscribe(onNext: { location in
        updateUserLocation(location)
    })
    .disposed(by: disposeBag)

skipwhile {}

'

클로저 속의 내용을 만족할때 까지!!! 스킵. 첫 false가 나오면 그때부터 모두 통과

// 로그인 완료될 때까지 사용자 액션 무시
userActions
    .skipWhile { _ in !authService.isLoggedIn }
    .subscribe(onNext: { action in
        processUserAction(action)
    })
    .disposed(by: disposeBag)

.skipUntil

 

다른 Observable(트리거)이 이벤트를 방출할 때까지 소스 Observable의 모든 이벤트를 건너뛰는 연산자입니다.

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

source
    .skipUntil(trigger)
    .subscribe(onNext: { value in
        print(value)
    })
    .disposed(by: disposeBag)

source.onNext("A") // 무시
source.onNext("B") // 무시

trigger.onNext(()) // 트리거 발생

source.onNext("C") // 출력: C
source.onNext("D") // 출력: D

흠...이건 언제 쓸수있을까요?

// 앱이 foreground로 돌아온 후에만 위치 업데이트 처리
locationManager.rx.location
    .skipUntil(NotificationCenter.default.rx.notification(UIApplication.didBecomeActiveNotification))
    .subscribe(onNext: { location in
        updateMapWithLocation(location)
    })
    .disposed(by: disposeBag)

특정 이벤트(권환획득, 로그인 성공)이 발생한 이후에만 데이터를 처리해야 할 때 유용

3. Taking Operator

: 어떠한 조건에 맞는 값들만 가져와 방출한다. Skipping과는 반대.

Take(count)

처음으로 부터 2개만 가져오겟다.

// 처음 5개의 검색 결과만 표시
searchResults
    .take(5)
    .bind(to: quickResultsView.rx.items)
    .disposed(by: disposeBag)

TakeWhile

SkipWhile과 비슷하면서 다르네... 조건을 만나기전까지 이벤트값만 방출하고 그 뒤론 무시한다.

// 오류가 발생하기 전까지의 응답만 처리
apiResponses
    .takeWhile { !$0.hasError }
    .subscribe(onNext: { response in
        processValidResponse(response)
    })
    .disposed(by: disposeBag)

TakeUntil

다른 옵저버블 실행이 있기전까지만 방출하고 그 뒤론 종료

// ViewController가 해제될 때 자동으로 구독 종료
apiClient.fetchData()
    .takeUntil(self.rx.deallocated)
    .subscribe(onNext: { data in
        self.updateUI(with: data)
    })
    .disposed(by: disposeBag)

이러면 disposbag이 필요 없겟네.

4. Distinct Operator

Distinct 연산자는 시퀀스에서 중복되는 요소를 필터링하여 유니크한 요소만 방출합니다.

distinctUntilChanged 

연속적으로 중복되는 요소를 필터링합니다. 이전 요소와 같은 요소만 필터링되며, 이전에 나온 적이 있지만 연속되지 않은 요소는 통과됩니다.

// 불필요한 API 요청 줄이기
searchBar.rx.text.orEmpty
    .distinctUntilChanged()
    .flatMapLatest { query -> Observable<[SearchResult]> in
        return api.search(query: query)
    }
    .bind(to: tableView.rx.items)
    .disposed(by: disposeBag)

distinctUntilChanged(_:)

커스텀 비교 로직을 사용하여 연속적인 중복을 필터링합니다. 복잡한 객체나 특정 속성만 비교해야 하는 경우에 유용합니다.

// 위치가 유의미하게 변경될 때만 지도 업데이트
locationManager.rx.location
  .distinctUntilChanged { previous, current in
    // 10미터 이내 변화는 무시
    return previous.distance(from: current) < 10.0
  }
  .subscribe(onNext: { location in
    mapView.centerToLocation(location)
  })
  .disposed(by: disposeBag)

실전 활용

1. 검색 기능 최적화 

searchBar.rx.text.orEmpty
	.debounce(.mileseconds(300), scheduler: MainScheduler.instance) // 타이핑 완료를 기다림
    .distinctUntilChanged() //중복요청방지
    .filter {!$0.isEmpty} //빈 검색어 필터링
    .skip(1) //초기이벤트 무시
    .flatMapLatest { [weak self] query -> Observable<[SearchResult]> in
        guard let self = self else { return .just([]) }

        self.activityIndicator.startAnimating()
        return self.searchService.search(query: query)
          .do(onNext: { _ in
            self.activityIndicator.stopAnimating()
          }, onError: { _ in
            self.activityIndicator.stopAnimating()
          })
          .catchAndReturn([])
      }
      .bind(to: tableView.rx.items(cellIdentifier: "ResultCell")) { index, model, cell in
        cell.configure(with: model)
      }
      .disposed(by: disposeBag)

 

2. 사용자 위치기반 알림

// 위치 업데이트 처리 최적화
locationManager.rx.location
  .skip(3) // 초기 부정확한 위치 무시
  .distinctUntilChanged { prev, current in
    // 50미터 미만 이동은 무시
    return prev.distance(from: current) < 50
  }
  .throttle(.seconds(60), scheduler: MainScheduler.instance) // 과도한 업데이트 방지
  .flatMapLatest { [weak self] location -> Observable<[Notification]> in
    guard let self = self else { return .empty() }
    
    return self.notificationService
      .checkNearbyNotifications(at: location)
      .take(5) // 최대 5개 알림만 표시
      .takeUntil(self.appStateService.didEnterBackground) // 백그라운드 진입 시 중단
  }
  .filter { !$0.isEmpty } // 알림이 있을 때만 처리
  .subscribe(onNext: { notifications in
    showLocalNotifications(notifications)
  })
  .disposed(by: disposeBag)

성능을 위해서는 연산자 순서 최적화도 중요할것같다.

 

Share

Observable 시퀀스를 여러 Observer가 공유할 수 있도록 해주는 오퍼레이터입니다. 이 오퍼레이터는 여러 구독자가 하나의 스트림을 공유하도록 하여 중복 실행을 방지합니다.'

기본적으로 RxSwift에서 Observable은 각 구독마다 새로운 시퀀스를 실행합니다. 이는 API 호출이나 무거운 연산을 포함하는 스트림의 경우 성능 문제를 일으킬 수 있습니다. share()는 이런 문제를 해결해줍니다.

// share() 사용 전 - API 호출이 두 번 발생
let observable = apiClient.fetchData()

observable
    .subscribe(onNext: { data in
        print("첫 번째 구독: \(data)")
    })
    .disposed(by: disposeBag)

observable
    .subscribe(onNext: { data in
        print("두 번째 구독: \(data)")
    })
    .disposed(by: disposeBag)

// share() 사용 후 - API 호출이 한 번만 발생
let sharedObservable = apiClient.fetchData().share()

sharedObservable
    .subscribe(onNext: { data in
        print("첫 번째 구독: \(data)")
    })
    .disposed(by: disposeBag)

sharedObservable
    .subscribe(onNext: { data in
        print("두 번째 구독: \(data)")
    })
    .disposed(by: disposeBag)

주의사항:

1. share()는 구독이 시작된 후 발생하는 이벤트만 공유합니다. 이미 발생한 이벤트는 새 구독자에게 전달되지 않습니다.
2. share(replay:scope:)를 사용할 때 메모리 관리에 주의해야 합니다. replay 값이 크면 메모리 사용량이 증가할 수 있습니다.
3. scope: .forever는 구독자가 없어도 이벤트를 캐시하므로 메모리 누수에 주의해야 합니다.

언제 사용해야 할까요?:

• 동일한 데이터를 여러 Observer가 구독할 때
• 네트워크 요청이나 무거운 연산을 중복 실행하지 않아야 할 때
• 최근 결과를 새 구독자에게 즉시 제공해야 할 때(share(replay:))

Transforming Operators

이전에는 이제 필터링을 통해 원하는 시퀀스의 요소만 작업을 해주었다면 이번에는 시퀀스의 요소를 다른 형태로 조작하거나 변환하는 연산자입니다.

toArray()

개별이벤트들을 배열로 변환

// 여러 페이지의 결과를 하나로 모으기
func loadAllItems() -> Observable<[Item]> {
    return Observable.range(start: 1, count: 3)
        .flatMap { page in
            return apiClient.fetchItems(page: page)
        }
        .toArray()
}

페이지네이션된 결과를 하나의 배열로 모을 때 유용하다.

flatMap(_:)

Map과 공통점은 각 요소를 새로운 Observable로 변환하지만 flatMap은 하나의 Observable로 평탄화해준다.

// 사용자 정보를 가져온 후 해당 사용자의 친구 목록 가져오기
authService.currentUser()
    .flatMap { user -> Observable<[Friend]> in
        return friendsAPI.getFriends(for: user.id)
    }
    .subscribe(onNext: { friends in
        self.updateFriendsList(friends)
    })
    .disposed(by: disposeBag)

중첩된 API호출할때 유용할 것같다.

flatMapLatest

flatMap과 유사하지만 새로운 Observable이 생성될때마다 이전 Observable구독을 해제한다.

searchBar.rx.text.orEmpty
    .debounce(.milliseconds(300), scheduler: MainScheduler.instance)
    .flatMapLatest { query -> Observable<[SearchResult]> in
        return searchAPI.search(query: query)
    }
    .subscribe(onNext: { results in
        self.updateResults(results)
    })
    .disposed(by: disposeBag)

검색기능구현에서 사용자가 새로운 검색어를 입력할때마다 이전 검색 요청을 취소하고 새로운 요청만 처리하는데 유용할것같다.

CompactMap(_:)

nil 결과를 제외하고 유효한 값만 변환후 통과 -> Optinoal 값을 처리할때 유용

scan(_:accumulator:)

이전값과 현재 값을 결합하여 누적 결과를 생성