Make pausableBuffered robust to reentrancy (#159)

Author: matthewvalentine

Source/RxSwift/pausableBuffered.swift

@@ -35,24 +35,29 @@ extension ObservableType {
             }
 
             var paused = true
+            var flushIndex = 0
             let lock = NSRecursiveLock()
 
             let flush = {
-                for value in buffer {
-                    observer.onNext(value)
+                while flushIndex < buffer.count {
+                    flushIndex += 1
+                    observer.onNext(buffer[flushIndex - 1])
+                }
+                if buffer.count > 0 {
+                    flushIndex = 0
+                    buffer.removeAll(keepingCapacity: limit != nil)
                 }
-                buffer.removeAll(keepingCapacity: limit != nil)
             }
 
-            let boundaryDisposable = pauser.subscribe { event in
+            let boundaryDisposable = pauser.distinctUntilChanged(==).subscribe { event in
                 lock.lock(); defer { lock.unlock() }
                 switch event {
                 case .next(let resume):
-                    paused = !resume
-
                     if resume && buffer.count > 0 {
                         flush()
                     }
+                    paused = !resume
+
                 case .completed:
                     observer.onCompleted()
                 case .error(let error):

Tests/RxSwift/pausableBufferedTests.swift

@@ -184,4 +184,96 @@ class PausableBufferedTests: XCTestCase {
             Subscription(200, 230)
             ])
     }
+
+    func testPausedReentrantPauser() {
+        let underlying = scheduler.createHotObservable([
+            next(110, 1),
+            next(210, 2),
+            next(310, 3),
+            next(410, 4),
+            completed(500),
+            ])
+
+        let pauser = scheduler.createHotObservable([
+            next(301, true),
+            ])
+
+        let res = scheduler.start(disposed: 1000) { () -> Observable<Int> in
+            return Observable.create { observer in
+                let pauserSubject = PublishSubject<Bool>()
+                let allPausers = Observable.merge(
+                    pauser.asObservable(),
+                    pauserSubject.asObservable())
+
+                let buffered = underlying
+                    .pausableBuffered(allPausers, limit: nil)
+                    .share()
+
+                let pauserEcho = buffered
+                    .take(2)
+                    .map { _ in true }
+                    .bind(to: pauserSubject)
+
+                return Disposables.create(pauserEcho, buffered.subscribe(observer))
+            }
+        }
+
+        XCTAssertEqual(res.events, [
+            next(301, 2),
+            next(310, 3),
+            next(410, 4),
+            completed(500),
+            ])
+
+        XCTAssertEqual(underlying.subscriptions, [
+            Subscription(200, 500),
+            ])
+    }
+
+    func testPausedReentrantUnderlying() {
+        let underlying = scheduler.createHotObservable([
+            next(210, 1),
+            next(210, 2),
+            next(210, 3),
+            next(310, 4),
+            completed(500),
+            ])
+
+        let pauser = scheduler.createHotObservable([
+            next(301, true),
+            ])
+
+        let res = scheduler.start(disposed: 1000) { () -> Observable<Int> in
+            return Observable.create { observer in
+                let underlyingSubject = PublishSubject<Int>()
+                let allUnderlying = Observable.merge(
+                    underlying.asObservable(),
+                    underlyingSubject.asObservable())
+
+                let buffered = allUnderlying
+                    .pausableBuffered(pauser, limit: nil)
+                    .share()
+
+                let underlyingEcho = buffered
+                    .take(2)
+                    .bind(to: underlyingSubject)
+
+                return Disposables.create(underlyingEcho, buffered.subscribe(observer))
+            }
+        }
+
+        XCTAssertEqual(res.events, [
+            next(301, 1),
+            next(301, 2),
+            next(301, 3),
+            next(301, 1),
+            next(301, 2),
+            next(310, 4),
+            completed(500),
+            ])
+
+        XCTAssertEqual(underlying.subscriptions, [
+            Subscription(200, 500),
+            ])
+    }
 }