RxSwift - 如何节制缓冲区的时间跨度

Question

我正在尝试重新创建一个代码片段，该代码片段基本上统计一个按钮在一行中被单击的次数。代码是在RxJS中，我试图将它转换为RxSwift用于学习目的，但是可以找出缓冲区和节流部分。

You can see the js code on jsfiddle

目前我有这个

tapButton.rx.tap  
    .buffer(timeSpan: 0.25, count: 10, scheduler: MainScheduler.instance) 
    .map {$0.count} 
    .filter { $0 >= 2 } 
    .subscribe(onNext: { events in 
    print(events) 
    }).addDisposableTo(disposeBag) 

而且我想不通我怎么能延迟，直到攻两端并收集所有的值，因为像RxJS示例中的最后发射。

Answer 1

您遇到的问题是因为RxSwift buffer操作员不像RxJS buffer操作员那样工作。它更像RxJS bufferWithTimeOrCount运营商。

目前，从版本3.4.0开始，没有等效于buffer运算符。它的签名会像func buffer(_ boundary: Observer<BoundaryType>) -> Observable<[E]>

这是一个有趣的问题来回答。我最终在这个答案的底部提供了一个缓冲区操作符。这里是出了在Andre的代码I的定义将如何写溶液：

let trigger = button.rx.tap.debounce(0.25, scheduler: MainScheduler.instance) 
    let clickStream = button.rx.tap.asObservable() 
     .buffer(trigger) 
     .map { $0.count } 
     .map { $0 == 1 ? "click" : "\($0)x clicks" } 

    let clearStream = clickStream 
     .debounce(10.0, scheduler: MainScheduler.instance) 
     .map { _ in "" } 

    Observable.merge([clickStream, clearStream]) 
     .bind(to: label.rx.text) 
     .disposed(by: bag) 

上述代码应放置在视图控制器的viewDidLoad方法。我做了一个大的改变和一个小的改变。这个小小的变化是我用反弹而不是油门。再一次，我认为RxJS的油门与RxSwift的油门不同。最大的改变是我将multiClickStream和singleClickStream结合起来。我不完全确定他为什么做了两个独立的流...

我做的另一个改变是将影响标签的所有可观察物卷成标签可以绑定的一个可观察物，而不是具有不同的可观察物。我认为这更清洁。

下面是我定义的缓冲区操作符。

extension Observable { 

    /// collects elements from the source sequence until the boundary sequence fires. Then it emits the elements as an array and begins collecting again. 
    func buffer<U>(_ boundary: Observable<U>) -> Observable<[E]> { 
     return Observable<[E]>.create { observer in 
      var buffer: [E] = [] 
      let lock = NSRecursiveLock() 
      let boundaryDisposable = boundary.subscribe { event in 
       lock.lock(); defer { lock.unlock() } 
       switch event { 
       case .next: 
        observer.onNext(buffer) 
        buffer = [] 
       default: 
        break 
       } 
      } 
      let disposable = self.subscribe { event in 
       lock.lock(); defer { lock.unlock() } 
       switch event { 
       case .next(let element): 
        buffer.append(element) 
       case .completed: 
        observer.onNext(buffer) 
        observer.onCompleted() 
       case .error(let error): 
        observer.onError(error) 
        buffer = [] 
       } 
      } 
      return Disposables.create([disposable, boundaryDisposable]) 
     } 
    } 
}