Custom Combine RateLimitedScheduler

Я пытаюсь реализовать планировщик ограничения скорости в сочетании. Документы плохо подходят для Scheduler, поэтому я делаю некоторые предположения о том, как они работают внутри. Я предполагаю, что функция schedule вызывается для каждого события в потоке, но это не то, что я наблюдаю на практике. Я вижу только пару вызовов этой функции в потоке многих других событий.

Вот моя реализация, которая сейчас практически ничего не делает и позволяет всем событиям проходить ...

public class RateLimitedScheduler: Scheduler {
    public typealias SchedulerTimeType = DispatchQueue.SchedulerTimeType
    public typealias SchedulerOptions = DispatchQueue.SchedulerOptions
    
    // MARK: API
    
    public var now: SchedulerTimeType { queue.now }
    public var minimumTolerance: SchedulerTimeType.Stride { queue.minimumTolerance }

    public func schedule(options: SchedulerOptions? = nil, _ action: @escaping () -> Void) {
        queue.schedule(after: nextDeadline(), tolerance: .init(floatLiteral: period / 10), options: options, action)
    }

    public func schedule(after date: SchedulerTimeType, tolerance: SchedulerTimeType.Stride, options: SchedulerOptions? = nil, _ action: @escaping () -> Void) {
        let deadline = max(nextDeadline(), date)
        queue.schedule(after: deadline, tolerance: tolerance, options: options, action)
    }

    public func schedule(after date: SchedulerTimeType,
                  interval: SchedulerTimeType.Stride,
                  tolerance: SchedulerTimeType.Stride,
                  options: SchedulerOptions? = nil,
                  _ action: @escaping () -> Void) -> Cancellable {
        let deadline = max(nextDeadline(), date)
        return queue.schedule(after: deadline, interval: interval, tolerance: tolerance, options: options, action)
    }
    
    // MARK: Initialization
    
    public init(maxEvents: Int, period: TimeInterval, queue: DispatchQueue = .main) {
        self.maxEvents = maxEvents
        self.period = period
        self.queue = queue
    }
    
    // MARK: Private
    
    private let maxEvents: Int
    private let period: TimeInterval
    private let queue: DispatchQueue

    private var eventCount = 0
    private var windowStartTime = DispatchTime.now()

    private func nextDeadline() -> SchedulerTimeType {
        let now = DispatchTime.now()

        if eventCount < maxEvents {
            eventCount += 1
            return SchedulerTimeType(max(windowStartTime, now))
        }

        let nextStartTime = windowStartTime + period
        eventCount = 1
        
        if now > nextStartTime {
            windowStartTime = now
            return SchedulerTimeType(max(windowStartTime, now))
        }
        
        windowStartTime = nextStartTime
            
        return SchedulerTimeType(max(nextStartTime, now))
    }
}

И для проверки:

let cancellable = Array(repeating: 0, count: 20).publisher
    .map { $0 * 1 }
    .subscribe(on: RateLimitedScheduler(maxEvents: 5, period: 1))
    .sink(receiveValue: { _ in
        print(Date())
    })

Мой алгоритм для nextDeadline определенно работает, поскольку я тестировал его вне планировщика, так что проблема не в этом.

Итак, мои вопросы:

1. Верно ли мое предположение, что каждое событие должно запускать функцию расписания в планировщике? И если нет, то почему?
2. Может ли кто-нибудь помочь разобраться в этом или предоставить мне альтернативу?