[Swift] RxSwift의 Throttle과 Combine의 Throttle은 다르다(부제: 버튼 중복 클릭 해결)

서론

유저가 버튼을 터치할 때 중복으로 터치하는 경우(일명 따닥..)가 생겨 RxSwift의 throttle operator를 사용해서 짧은 시간(주로 0.3~0.5초) 내에 터치가 연속적으로 들어오면 첫번째 터치 이외에는 입력을 무시했습니다.

그런데 최근에 RxSwift에서 Combine를 옮기는 작업을 하는 도중 Combine에도 비슷한 인터페이스를 갖고 있는 throttle operator를 확인하고 적용했습니다.

적용해보고 확인해보니 제가 원하는 동작과는 달랐습니다. 혹시나 저만 이런가 싶어 검색해보니 다른 커뮤니티에서도 저와 같은 이슈를 겪고 있었고, 따닥을 해결할 수 있는 Custom Operator를 추가하기로 합니다.

본론

일단 Combine framework의 throttle operator의 인터페이스를 확인해봅시다.

RxSwift와 유사하게 interval, scheduler, latest 총 3가지 인자를 받고 있는 걸 확인할 수 있습니다. 저도 유사하게 만들어보겠습니다. rx_throttle 라는 이름으로 아래와 같이 확장해줍니다.

import Combine

extension Publisher {
    func rx_throttle<S>(
        for interval: S.SchedulerTimeType.Stride,
        scheduler: S,
        latest: Bool = true
    ) -> Publishers.RxThrottle<Self, S> where S: Scheduler {
        return Publishers.RxThrottle(
            upstream: self,
            interval: interval,
            scheduler: scheduler,
            latest: latest
        )
    }
}

이제 RxThrottle의 기능을 담당할 Custom Publisher를 만들어줍니다.

extension Publishers {
    struct RxThrottle<Upstream, SchedulerType>: Publisher where Upstream: Publisher, SchedulerType: Scheduler {
    // ..
    }
}

대부분의 Publisher는 struct로 만들게 됩니다. 물론 아닌 경우도 있긴 한데, 비동기 상태에서의 상태 저장과 같이 특수한 경우가 아닐 경우 struct로 객체를 선언해주고 Publisher protocol를 conform 하면 됩니다.

Publisher 프로토콜을 conform 받았으면 Output, Failure, receive 함수를 구현해줍니다.

extension Publishers {
    struct RxThrottle<Upstream, SchedulerType>: Publisher where Upstream: Publisher, SchedulerType: Scheduler {
        typealias Output = Upstream.Output
        typealias Failure = Upstream.Failure
        
        let upstream: Upstream
        let interval: SchedulerType.SchedulerTimeType.Stride
        let scheduler: SchedulerType
        let latest: Bool
        
        init(upstream: Upstream, interval: SchedulerType.SchedulerTimeType.Stride, scheduler: SchedulerType, latest: Bool) {
            self.upstream = upstream
            self.interval = interval
            self.scheduler = scheduler
            self.latest = latest
        }
        
        func receive<S>(subscriber: S) where S : Subscriber, Upstream.Failure == S.Failure, Upstream.Output == S.Input {
            let subscription = RxThrottleSubscriber(
                subscriber: subscriber,
                interval: interval,
                scheduler: scheduler,
                latest: latest
            )
            upstream.subscribe(subscription)
        }
    }
}

이제 실질적인 로직이 들어갈 Subscriber을 구현해주어야 합니다. 저는 RxThrottleSubscriber라는 이름으로 구현하도록 하겠습니다.

final class RxThrottleSubscriber<Downstream: Subscriber, SchedulerType>: Subscriber where SchedulerType: Scheduler {
    // ..
}

위 처럼 객체를 만들어주고 Subscriber 프로토콜이 요구하는 아래의 기능들을 구현합니다.

기능 구현은 아래와 같이 합니다.

final class RxThrottleSubscriber<Downstream: Subscriber, SchedulerType>: Subscriber where SchedulerType: Scheduler {
    typealias Input = Downstream.Input
    typealias Failure = Downstream.Failure
    
    private var downstream: Downstream? = nil
    private let interval: SchedulerType.SchedulerTimeType.Stride
    private let scheduler: SchedulerType
    private let latest: Bool
    private var previousValue: Input? = nil
    private var isRequesting: Bool = false
    private var requestedDemand: Subscribers.Demand = .none
    private var throttleTimer: Cancellable? = nil
    
    init(
        subscriber: Downstream,
        interval: SchedulerType.SchedulerTimeType.Stride,
        scheduler: SchedulerType,
        latest: Bool
    ) {
        self.downstream = subscriber
        self.interval = interval
        self.scheduler = scheduler
        self.latest = latest
    }
    
    func receive(subscription: Subscription) {
        subscription.request(.unlimited)
    }
    
    func receive(_ input: Downstream.Input) -> Subscribers.Demand {
        previousValue = input
        startThrottleTimer(input)
        return .none
    }
    
    func receive(completion: Subscribers.Completion<Downstream.Failure>) {
        throttleTimer?.cancel()
        downstream?.receive(completion: completion)
    }
    
    func cancel() {
        throttleTimer?.cancel()
        throttleTimer = nil
        downstream = nil
    }
    
    private func startThrottleTimer(_ input: Downstream.Input) {
        guard !isRequesting else { return }
        
        isRequesting = true

        if latest {
            if throttleTimer == nil {
                _ = downstream?.receive(input)
                
                throttleTimer = scheduler.schedule(after: scheduler.now.advanced(by: interval), interval: interval, tolerance: .zero, options: nil) { [weak self] in
                    if let previousValue = self?.previousValue {
                        _ = self?.downstream?.receive(previousValue)
                        self?.previousValue = nil
                    } else {
                        self?.throttleTimer?.cancel()
                        self?.throttleTimer = nil
                    }
                    
                    self?.isRequesting = false
                }
            }
        } else {
            _ = downstream?.receive(input)
            
            scheduler.schedule(after: scheduler.now.advanced(by: interval), tolerance: .zero, options: nil) { [weak self] in
                self?.isRequesting = false
            }
        }
    }
}

이제 아래와 같이 호출해서 사용하면 됩니다.

publisher
     .rx_throttle(for: .milliseconds(400), scheduler: DispatchQueue.main, latest: false)
     .sink { _ in
         // ...
     }
     .store(in: &cancellables)

이제 버튼 따닥을 해결할 수 있게 되었습니다.

전체 소스 코드

결론

인터페이스가 비슷하게 생겼고 이름이 같다고 해도 돌다리도 두들겨 보고 건너자.