swift - RxSwift - How to throttle buffer's time span -

i trying recreate code snippet counts how many times button clicked in row. code in rxjs , trying convert rxswift learning purposes can'f figure out buffer , throttle part.

you can see js code on jsfiddle

currently 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 can't figure out how can delay until tapping ends , collect values since last emission in rxjs example.

the problem having because rxswift buffer operator doesn't work rxjs buffer operator. works more rxjs bufferwithtimeorcount operator.

currently, of version 3.4.0, there no equivalent buffer operator. it's signature func buffer(_ boundary: observer<boundarytype>) -> observable<[e]>

this has been fun question answer. ended making buffer operator provide @ bottom of answer. here how write solution out 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 placed in view controller's viewdidload method. there 1 big change , 1 small change made. small change used debounce instead of throttle. again, think rxjs's throttle works differently rxswift's throttle does. big change combined multiclickstream , singleclickstream. i'm not entirely sure why made 2 separate streams...

another change made roll observables affect label 1 observable label bind to, instead of having different ones. think cleaner.

below buffer operator defined.

extension observable {      /// collects elements source sequence until boundary sequence fires. emits elements array , 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])         }     } }