这是一个有趣的问题。我玩过Timer.publish、buffer、zip 和throttle 的各种组合，但我无法让任何组合完全按照你想要的方式工作。所以让我们编写一个自定义订阅者。
我们真正想要的是一个 API，当我们从上游获取输入时，我们还能够控制上游何时传递下一个输入。像这样的：
extension Publisher {
/// Subscribe to me with a stepping function.
/// - parameter stepper: A function I'll call with each of my inputs, and with my completion.
/// Each time I call this function with an input, I also give it a promise function.
/// I won't deliver the next input until the promise is called with a `.more` argument.
/// - returns: An object you can use to cancel the subscription asynchronously.
func step(with stepper: @escaping (StepEvent<Output, Failure>) -> ()) -> AnyCancellable {
???
}
}
enum StepEvent<Input, Failure: Error> {
/// Handle the Input. Call `StepPromise` when you're ready for the next Input,
/// or to cancel the subscription.
case input(Input, StepPromise)
/// Upstream completed the subscription.
case completion(Subscribers.Completion<Failure>)
}
/// The type of callback given to the stepper function to allow it to continue
/// or cancel the stream.
typealias StepPromise = (StepPromiseRequest) -> ()
enum StepPromiseRequest {
// Pass this to the promise to request the next item from upstream.
case more
// Pass this to the promise to cancel the subscription.
case cancel
}
有了这个step API，我们可以编写一个pace 操作符来做你想做的事：
extension Publisher {
func pace<Context: Scheduler, MySubject: Subject>(
_ pace: Context.SchedulerTimeType.Stride, scheduler: Context, subject: MySubject)
-> AnyCancellable
where MySubject.Output == Output, MySubject.Failure == Failure
{
return step {
switch $0 {
case .input(let input, let promise):
// Send the input from upstream now.
subject.send(input)
// Wait for the pace interval to elapse before requesting the
// next input from upstream.
scheduler.schedule(after: scheduler.now.advanced(by: pace)) {
promise(.more)
}
case .completion(let completion):
subject.send(completion: completion)
}
}
}
}
这个pace 运算符采用pace（输出之间所需的间隔）、一个调度事件的调度程序和一个重新发布上游输入的subject。它通过subject 发送每个输入来处理每个输入，然后使用调度程序等待步调间隔，然后再从上游请求下一个输入。
现在我们只需要实现step 操作符。在这里，Combine 并没有给我们太多帮助。它确实有一个称为“背压”的功能，这意味着发布者无法向下游发送输入，直到下游通过向上游发送 Subscribers.Demand 来请求它。通常你会看到下游向上游发送.unlimited 需求，但我们不会这样做。相反，我们将利用背压。在 stepper 完成 promise 之前，我们不会向上游发送任何需求，然后我们只会发送 .max(1) 的需求，因此我们使上游与 stepper 同步运行。 （我们还必须发送.max(1) 的初始需求才能开始整个过程。）
好的，所以需要实现一个接受步进函数并符合Subscriber 的类型。查看Reactive Streams JVM Specification 是个好主意，因为Combine 是基于该规范。
让实现变得困难的是，有几件事可以异步调用我们的订阅者：
- 上游可以从任何线程调用订阅者（但需要序列化其调用）。
- 在我们为步进器提供了 Promise 函数后，步进器可以在任何线程上调用这些 Promise。
- 我们希望订阅可以取消，并且可以在任何线程上取消。
- 所有这些异步性意味着我们必须用锁来保护我们的内部状态。
- 我们必须小心不要在持有该锁时调用，以避免死锁。
我们还将通过为每个 Promise 分配一个唯一的 id 来保护订户免受涉及重复调用 Promise 或调用过时 Promise 的恶作剧。
这是我们的基本订阅者定义：
import Combine
import Foundation
public class SteppingSubscriber<Input, Failure: Error> {
public init(stepper: @escaping Stepper) {
l_state = .subscribing(stepper)
}
public typealias Stepper = (Event) -> ()
public enum Event {
case input(Input, Promise)
case completion(Completion)
}
public typealias Promise = (Request) -> ()
public enum Request {
case more
case cancel
}
public typealias Completion = Subscribers.Completion<Failure>
private let lock = NSLock()
// The l_ prefix means it must only be accessed while holding the lock.
private var l_state: State
private var l_nextPromiseId: PromiseId = 1
private typealias PromiseId = Int
private var noPromiseId: PromiseId { 0 }
}
请注意，我将之前的辅助类型（StepEvent、StepPromise 和 StepPromiseRequest）移至 SteppingSubscriber 并缩短了它们的名称。
现在让我们考虑l_state 的神秘类型State。我们的订阅者可能处于哪些不同的状态？
- 我们可能正在等待从上游接收
Subscription 对象。
- 我们可能已经收到来自上游的
Subscription 并正在等待信号（来自上游的输入或完成，或者来自步进器的承诺的完成）。
- 我们可能会调用步进器，我们要小心，以防它在我们调用它时完成承诺。
- 我们可能已被取消或已收到来自上游的完成。
这是我们对State的定义：
extension SteppingSubscriber {
private enum State {
// Completed or cancelled.
case dead
// Waiting for Subscription from upstream.
case subscribing(Stepper)
// Waiting for a signal from upstream or for the latest promise to be completed.
case subscribed(Subscribed)
// Calling out to the stopper.
case stepping(Stepping)
var subscription: Subscription? {
switch self {
case .dead: return nil
case .subscribing(_): return nil
case .subscribed(let subscribed): return subscribed.subscription
case .stepping(let stepping): return stepping.subscribed.subscription
}
}
struct Subscribed {
var stepper: Stepper
var subscription: Subscription
var validPromiseId: PromiseId
}
struct Stepping {
var subscribed: Subscribed
// If the stepper completes the current promise synchronously with .more,
// I set this to true.
var shouldRequestMore: Bool
}
}
}
由于我们使用NSLock（为简单起见），让我们定义一个扩展来确保我们始终匹配锁定和解锁：
fileprivate extension NSLock {
@inline(__always)
func sync<Answer>(_ body: () -> Answer) -> Answer {
lock()
defer { unlock() }
return body()
}
}
现在我们准备好处理一些事件了。最容易处理的事件是异步取消，这是Cancellable 协议的唯一要求。如果我们处于除.dead 之外的任何状态，我们希望成为.dead，如果有上游订阅，请取消它。
extension SteppingSubscriber: Cancellable {
public func cancel() {
let sub: Subscription? = lock.sync {
defer { l_state = .dead }
return l_state.subscription
}
sub?.cancel()
}
}
请注意，当lock 被锁定时，我不想调用上游订阅的cancel 函数，因为lock 不是递归锁，我不想冒死锁的风险。对lock.sync 的所有使用都遵循将任何调用推迟到解锁之后的模式。
现在让我们实现Subscriber 协议要求。首先，让我们处理从上游接收Subscription。唯一应该发生的情况是当我们处于 .subscribing 状态时，但 .dead 也是可能的，在这种情况下我们只想取消上游订阅。
extension SteppingSubscriber: Subscriber {
public func receive(subscription: Subscription) {
let action: () -> () = lock.sync {
guard case .subscribing(let stepper) = l_state else {
return { subscription.cancel() }
}
l_state = .subscribed(.init(stepper: stepper, subscription: subscription, validPromiseId: noPromiseId))
return { subscription.request(.max(1)) }
}
action()
}
请注意，在lock.sync 的这次使用中（以及以后的所有使用中），我返回了一个“动作”闭包，这样我就可以在锁被解锁后执行任意调用。
我们要解决的下一个Subscriber 协议要求是接收完成：
public func receive(completion: Subscribers.Completion<Failure>) {
let action: (() -> ())? = lock.sync {
// The only state in which I have to handle this call is .subscribed:
// - If I'm .dead, either upstream already completed (and shouldn't call this again),
// or I've been cancelled.
// - If I'm .subscribing, upstream must send me a Subscription before sending me a completion.
// - If I'm .stepping, upstream is currently signalling me and isn't allowed to signal
// me again concurrently.
guard case .subscribed(let subscribed) = l_state else {
return nil
}
l_state = .dead
return { [stepper = subscribed.stepper] in
stepper(.completion(completion))
}
}
action?()
}
对我们来说最复杂的Subscriber 协议要求是接收Input：
- 我们必须创建一个承诺。
- 我们必须将承诺传递给步进器。
- 步进器可以在返回之前完成承诺。
- stepper 返回后，我们要检查它是否完成了
.more 的promise，如果是，则向上游返回适当的需求。
由于我们必须在这项工作的中间调用步进器，所以我们有一些难看的 lock.sync 调用嵌套。
public func receive(_ input: Input) -> Subscribers.Demand {
let action: (() -> Subscribers.Demand)? = lock.sync {
// The only state in which I have to handle this call is .subscribed:
// - If I'm .dead, either upstream completed and shouldn't call this,
// or I've been cancelled.
// - If I'm .subscribing, upstream must send me a Subscription before sending me Input.
// - If I'm .stepping, upstream is currently signalling me and isn't allowed to
// signal me again concurrently.
guard case .subscribed(var subscribed) = l_state else {
return nil
}
let promiseId = l_nextPromiseId
l_nextPromiseId += 1
let promise: Promise = { request in
self.completePromise(id: promiseId, request: request)
}
subscribed.validPromiseId = promiseId
l_state = .stepping(.init(subscribed: subscribed, shouldRequestMore: false))
return { [stepper = subscribed.stepper] in
stepper(.input(input, promise))
let demand: Subscribers.Demand = self.lock.sync {
// The only possible states now are .stepping and .dead.
guard case .stepping(let stepping) = self.l_state else {
return .none
}
self.l_state = .subscribed(stepping.subscribed)
return stepping.shouldRequestMore ? .max(1) : .none
}
return demand
}
}
return action?() ?? .none
}
} // end of extension SteppingSubscriber: Publisher
我们的订阅者需要处理的最后一件事是完成一个承诺。这很复杂有几个原因：
- 我们希望防止承诺被多次完成。
- 我们希望防止旧承诺的完成。
- 承诺完成后，我们可以处于任何状态。
因此：
extension SteppingSubscriber {
private func completePromise(id: PromiseId, request: Request) {
let action: (() -> ())? = lock.sync {
switch l_state {
case .dead, .subscribing(_): return nil
case .subscribed(var subscribed) where subscribed.validPromiseId == id && request == .more:
subscribed.validPromiseId = noPromiseId
l_state = .subscribed(subscribed)
return { [sub = subscribed.subscription] in
sub.request(.max(1))
}
case .subscribed(let subscribed) where subscribed.validPromiseId == id && request == .cancel:
l_state = .dead
return { [sub = subscribed.subscription] in
sub.cancel()
}
case .subscribed(_):
// Multiple completion or stale promise.
return nil
case .stepping(var stepping) where stepping.subscribed.validPromiseId == id && request == .more:
stepping.subscribed.validPromiseId = noPromiseId
stepping.shouldRequestMore = true
l_state = .stepping(stepping)
return nil
case .stepping(let stepping) where stepping.subscribed.validPromiseId == id && request == .cancel:
l_state = .dead
return { [sub = stepping.subscribed.subscription] in
sub.cancel()
}
case .stepping(_):
// Multiple completion or stale promise.
return nil
}
}
action?()
}
}
哇！
完成所有这些后，我们就可以编写真正的step 运算符了：
extension Publisher {
func step(with stepper: @escaping (SteppingSubscriber<Output, Failure>.Event) -> ()) -> AnyCancellable {
let subscriber = SteppingSubscriber<Output, Failure>(stepper: stepper)
self.subscribe(subscriber)
return .init(subscriber)
}
}
然后我们可以从上面尝试pace 运算符。由于我们在SteppingSubscriber 中不做任何缓冲，而且上游通常没有缓冲，我们将在上游和pace 运算符之间添加一个buffer。
var cans: [AnyCancellable] = []
func application(_ application: UIApplication, didFinishLaunchingWithOptions launchOptions: [UIApplication.LaunchOptionsKey: Any]?) -> Bool {
let erratic = Just("A").delay(for: 0.0, tolerance: 0.001, scheduler: DispatchQueue.main).eraseToAnyPublisher()
.merge(with: Just("B").delay(for: 0.3, tolerance: 0.001, scheduler: DispatchQueue.main).eraseToAnyPublisher())
.merge(with: Just("C").delay(for: 0.6, tolerance: 0.001, scheduler: DispatchQueue.main).eraseToAnyPublisher())
.merge(with: Just("D").delay(for: 5.0, tolerance: 0.001, scheduler: DispatchQueue.main).eraseToAnyPublisher())
.merge(with: Just("E").delay(for: 5.3, tolerance: 0.001, scheduler: DispatchQueue.main).eraseToAnyPublisher())
.merge(with: Just("F").delay(for: 5.6, tolerance: 0.001, scheduler: DispatchQueue.main).eraseToAnyPublisher())
.handleEvents(
receiveOutput: { print("erratic: \(Double(DispatchTime.now().rawValue) / 1_000_000_000) \($0)") },
receiveCompletion: { print("erratic: \(Double(DispatchTime.now().rawValue) / 1_000_000_000) \($0)") }
)
.makeConnectable()
let subject = PassthroughSubject<String, Never>()
cans += [erratic
.buffer(size: 1000, prefetch: .byRequest, whenFull: .dropOldest)
.pace(.seconds(1), scheduler: DispatchQueue.main, subject: subject)]
cans += [subject.sink(
receiveCompletion: { print("paced: \(Double(DispatchTime.now().rawValue) / 1_000_000_000) \($0)") },
receiveValue: { print("paced: \(Double(DispatchTime.now().rawValue) / 1_000_000_000) \($0)") }
)]
let c = erratic.connect()
cans += [AnyCancellable { c.cancel() }]
return true
}
最后，这里是输出：
erratic: 223394.17115897 A
paced: 223394.171495405 A
erratic: 223394.408086369 B
erratic: 223394.739186984 C
paced: 223395.171615624 B
paced: 223396.27056174 C
erratic: 223399.536717127 D
paced: 223399.536782847 D
erratic: 223399.536834495 E
erratic: 223400.236808469 F
erratic: 223400.236886323 finished
paced: 223400.620542561 E
paced: 223401.703613078 F
paced: 223402.703828512 finished
- 时间戳以秒为单位。
- 不稳定的出版商的时间确实不稳定，有时甚至很接近。
- 即使发生不稳定事件的时间间隔不到一秒，起搏计时也始终相隔至少一秒。
- 当不稳定事件发生在前一个事件之后超过一秒时，起搏事件会在不稳定事件之后立即发送，不会再延迟。
- 有节奏的完成发生在最后一个有节奏的事件之后一秒，即使不稳定的完成发生在最后一个不稳定的事件之后。
buffer 不会发送完成，直到它在发送最后一个事件后收到另一个请求，并且该请求被起搏计时器延迟。
我已将 step 运算符的整个实现放在 this gist 中，以便于复制/粘贴。