RxSwift - How to throttle buffer's time span

Question

I am trying to recreate a code snippet that basically counts how many times a button was clicked in a row. The code is in RxJS and I am trying to convert it to RxSwift for learning purposes but can't figure out the buffer and throttle part.

You can see the js code on jsfiddle

Currently I have this

  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)

And I can't figure out how can I delay until tapping ends and collect all values since the last emission like in the RxJS example.

Answer 1

The problem you are having is because the RxSwift buffer operator doesn't work like the RxJS buffer operator. It works more like the RxJS bufferWithTimeOrCount operator.

Currently, as of version 3.4.0, there is no equivalent to the buffer operator. It's signature would be something like func buffer(_ boundary: Observer<BoundaryType>) -> Observable<[E]>

This has been a fun question to answer. I ended up making a buffer operator which I provide at the bottom of this answer. Here is how I would write the solution out that is defined in Andre's code:

    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)

The above code should be placed in the view controller's viewDidLoad method. There is one big change and one small change that I made. The small change is that I used debounce instead of throttle. Again, I think RxJS's throttle works differently than RxSwift's throttle does. The big change is that I combined his multiClickStream and singleClickStream. I'm not entirely sure why he made two separate streams...

Another change I made was to roll all the observables that affect the label into one observable that the label could bind to, instead of having different ones. I think this is cleaner.

Below is the buffer operator that I defined.

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])
        }
    }
}