Fixes issues with locking strategy for subjects. ReactiveX#936

kzaher · kzaher · commit c594e28fd94a · 2016-11-20T17:12:22.000+01:00
diff --git a/Rx.xcodeproj/project.pbxproj b/Rx.xcodeproj/project.pbxproj
@@ -676,6 +676,9 @@
 		C8941BE71BD56B0700A0E874 /* BlockingObservable+Operators.swift in Sources */ = {isa = PBXBuildFile; fileRef = C8941BE31BD56B0700A0E874 /* BlockingObservable+Operators.swift */; };
 		C89461751BC6C1210055219D /* ObservableConvertibleType.swift in Sources */ = {isa = PBXBuildFile; fileRef = C849BE2A1BAB5D070019AD27 /* ObservableConvertibleType.swift */; };
 		C89461761BC6C1220055219D /* ObservableConvertibleType.swift in Sources */ = {isa = PBXBuildFile; fileRef = C849BE2A1BAB5D070019AD27 /* ObservableConvertibleType.swift */; };
+		C89579381DE1FD8C00DC9C9C /* Anomalies.swift in Sources */ = {isa = PBXBuildFile; fileRef = C89579351DE1FD8000DC9C9C /* Anomalies.swift */; };
+		C895793A1DE1FD8D00DC9C9C /* Anomalies.swift in Sources */ = {isa = PBXBuildFile; fileRef = C89579351DE1FD8000DC9C9C /* Anomalies.swift */; };
+		C895793B1DE1FD8E00DC9C9C /* Anomalies.swift in Sources */ = {isa = PBXBuildFile; fileRef = C89579351DE1FD8000DC9C9C /* Anomalies.swift */; };
 		C89CDB361BCB0DD7002063D9 /* ShareReplay1.swift in Sources */ = {isa = PBXBuildFile; fileRef = C89CDB351BCB0DD7002063D9 /* ShareReplay1.swift */; };
 		C89CDB371BCB0DD7002063D9 /* ShareReplay1.swift in Sources */ = {isa = PBXBuildFile; fileRef = C89CDB351BCB0DD7002063D9 /* ShareReplay1.swift */; };
 		C89CDB381BCB0DD7002063D9 /* ShareReplay1.swift in Sources */ = {isa = PBXBuildFile; fileRef = C89CDB351BCB0DD7002063D9 /* ShareReplay1.swift */; };
@@ -1689,6 +1692,7 @@
 		C88FA53F1C25C4CC00CCFEA4 /* RxTests.framework */ = {isa = PBXFileReference; explicitFileType = wrapper.framework; includeInIndex = 0; path = RxTests.framework; sourceTree = BUILT_PRODUCTS_DIR; };
 		C8941BDE1BD5695C00A0E874 /* BlockingObservable.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = BlockingObservable.swift; sourceTree = "<group>"; };
 		C8941BE31BD56B0700A0E874 /* BlockingObservable+Operators.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = "BlockingObservable+Operators.swift"; sourceTree = "<group>"; };
+		C89579351DE1FD8000DC9C9C /* Anomalies.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; name = Anomalies.swift; path = Tests/RxSwiftTests/Anomalies.swift; sourceTree = SOURCE_ROOT; };
 		C89CDB351BCB0DD7002063D9 /* ShareReplay1.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; lineEnding = 0; path = ShareReplay1.swift; sourceTree = "<group>"; xcLanguageSpecificationIdentifier = xcode.lang.swift; };
 		C8A56AD71AD7424700B4673B /* RxSwift.framework */ = {isa = PBXFileReference; explicitFileType = wrapper.framework; includeInIndex = 0; path = RxSwift.framework; sourceTree = BUILT_PRODUCTS_DIR; };
 		C8B144FA1BD2D44500267DCE /* ConcurrentMainScheduler.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = ConcurrentMainScheduler.swift; sourceTree = "<group>"; };
@@ -2357,6 +2361,7 @@
 		C83509011C38706D0027C24C /* Tests */ = {
 			isa = PBXGroup;
 			children = (
+				C89579351DE1FD8000DC9C9C /* Anomalies.swift */,
 				C83509021C38706D0027C24C /* AnonymousObservable+Test.swift */,
 				C83509031C38706D0027C24C /* AssumptionsTest.swift */,
 				C83509041C38706D0027C24C /* BagTest.swift */,
@@ -3510,6 +3515,7 @@
 			files = (
 				033C2EF61D081C460050C015 /* UIScrollView+RxTests.swift in Sources */,
 				C835092A1C38706E0027C24C /* CLLocationManager+RxTests.swift in Sources */,
+				C89579381DE1FD8C00DC9C9C /* Anomalies.swift in Sources */,
 				C83509661C38706E0027C24C /* RxTest.swift in Sources */,
 				C83509611C38706E0027C24C /* ObserverTests.swift in Sources */,
 				C83509471C38706E0027C24C /* PrimitiveMockObserver.swift in Sources */,
@@ -3600,6 +3606,7 @@
 				C83509F31C38755D0027C24C /* AssumptionsTest.swift in Sources */,
 				C83509ED1C3875580027C24C /* MySubject.swift in Sources */,
 				C83509C61C3875220027C24C /* NSObject+RxTests.swift in Sources */,
+				C895793B1DE1FD8E00DC9C9C /* Anomalies.swift in Sources */,
 				C83509BB1C38750D0027C24C /* Control+RxTests.swift in Sources */,
 				C83509EE1C3875580027C24C /* Observable.Extensions.swift in Sources */,
 				C83509BD1C38750D0027C24C /* ControlPropertyTests.swift in Sources */,
@@ -3721,6 +3728,7 @@
 				C83509CC1C3875230027C24C /* NSLayoutConstraint+RxTests.swift in Sources */,
 				C8350A0C1C38755E0027C24C /* Observable+CreationTest.swift in Sources */,
 				C8350A1C1C38756B0027C24C /* Observable+StandardSequenceOperatorsTest.swift in Sources */,
+				C895793A1DE1FD8D00DC9C9C /* Anomalies.swift in Sources */,
 				C83509CA1C3875230027C24C /* Driver+Test.swift in Sources */,
 				C83509E41C3875580027C24C /* MockDisposable.swift in Sources */,
 				C83509D51C38753E0027C24C /* RxObjCRuntimeState.swift in Sources */,
diff --git a/RxSwift/Observables/Implementations/ShareReplay1.swift b/RxSwift/Observables/Implementations/ShareReplay1.swift
@@ -77,13 +77,13 @@ final class ShareReplay1<Element>
     }
 
     func on(event: Event<E>) {
-        _lock.lock(); defer { _lock.unlock() }
-        _synchronized_on(event)
+        _synchronized_on(event).on(event)
     }
 
-    func _synchronized_on(event: Event<E>) {
+    func _synchronized_on(_ event: Event<E>) -> Bag<AnyObserver<Element>> {
+        _lock.lock(); defer { _lock.unlock() }
         if _stopped {
-            return
+            return Bag()
         }
 
         switch event {
@@ -96,6 +96,6 @@ final class ShareReplay1<Element>
             _connection = nil
         }
 
-        _observers.on(event)
+        return _observers
     }
-}
+}
diff --git a/RxSwift/Observables/Implementations/ShareReplay1WhileConnected.swift b/RxSwift/Observables/Implementations/ShareReplay1WhileConnected.swift
@@ -71,22 +71,22 @@ final class ShareReplay1WhileConnected<Element>
     }
 
     func on(event: Event<E>) {
-        _lock.lock(); defer { _lock.unlock() }
-        _synchronized_on(event)
+        _synchronized_on(event).on(event)
     }
 
-    func _synchronized_on(event: Event<E>) {
+    func _synchronized_on(_ event: Event<E>) -> Bag<AnyObserver<Element>> {
+        _lock.lock(); defer { _lock.unlock() }
         switch event {
         case .Next(let element):
             _element = element
-            _observers.on(event)
+            return _observers
         case .Error, .Completed:
             _element = nil
             _connection?.dispose()
             _connection = nil
             let observers = _observers
             _observers = Bag()
-            observers.on(event)
+            return observers
         }
     }
-}
+}
diff --git a/RxSwift/Subjects/BehaviorSubject.swift b/RxSwift/Subjects/BehaviorSubject.swift
@@ -81,13 +81,13 @@ public final class BehaviorSubject<Element>
     - parameter event: Event to send to the observers.
     */
     public func on(event: Event<E>) {
-        _lock.lock(); defer { _lock.unlock() }
-        _synchronized_on(event)
+        _synchronized_on(event).on(event)
     }
 
-    func _synchronized_on(event: Event<E>) {
+    func _synchronized_on(event: Event<E>) -> Bag<AnyObserver<Element>> {
+        _lock.lock(); defer { _lock.unlock() }
         if _stoppedEvent != nil || _disposed {
-            return
+            return Bag()
         }
         
         switch event {
@@ -97,7 +97,7 @@ public final class BehaviorSubject<Element>
             _stoppedEvent = event
         }
         
-        _observers.on(event)
+        return _observers
     }
     
     /**
@@ -158,4 +158,4 @@ public final class BehaviorSubject<Element>
             _stoppedEvent = nil
         }
     }
-}
+}
diff --git a/RxSwift/Subjects/PublishSubject.swift b/RxSwift/Subjects/PublishSubject.swift
@@ -59,25 +59,29 @@ final public class PublishSubject<Element>
     - parameter event: Event to send to the observers.
     */
     public func on(event: Event<Element>) {
-        _lock.lock(); defer { _lock.unlock() }
-        _synchronized_on(event)
+        _synchronized_on(event).on(event)
     }
 
-    func _synchronized_on(event: Event<E>) {
+    func _synchronized_on(event: Event<E>) -> Bag<AnyObserver<Element>> {
+        _lock.lock(); defer { _lock.unlock() }
+
         switch event {
         case .Next(_):
             if _disposed || _stopped {
-                return
+                return Bag()
             }
             
-            _observers.on(event)
+            return _observers
         case .Completed, .Error:
             if _stoppedEvent == nil {
                 _stoppedEvent = event
                 _stopped = true
-                _observers.on(event)
+                let observers = _observers
                 _observers.removeAll()
+                return observers
             }
+
+            return Bag()
         }
     }
     
@@ -136,4 +140,4 @@ final public class PublishSubject<Element>
         _observers.removeAll()
         _stoppedEvent = nil
     }
-}
+}
diff --git a/RxSwift/Subjects/ReplaySubject.swift b/RxSwift/Subjects/ReplaySubject.swift
@@ -105,29 +105,30 @@ class ReplayBufferBase<Element>
     }
     
     override func on(event: Event<Element>) {
-        _lock.lock(); defer { _lock.unlock() }
-        _synchronized_on(event)
+        _synchronized_on(event).on(event)
     }
 
-    func _synchronized_on(event: Event<E>) {
+    func _synchronized_on(event: Event<E>) -> Bag<AnyObserver<Element>> {
+        _lock.lock(); defer { _lock.unlock() }
         if _disposed {
-            return
+            return Bag()
         }
         
         if _stoppedEvent != nil {
-            return
+            return Bag()
         }
         
         switch event {
         case .Next(let value):
             addValueToBuffer(value)
             trim()
-            _observers.on(event)
+            return _observers
         case .Error, .Completed:
             _stoppedEvent = event
             trim()
-            _observers.on(event)
+            let observers = _observers
             _observers.removeAll()
+            return observers
         }
     }
     
@@ -260,4 +261,4 @@ final class ReplayAll<Element> : ReplayManyBase<Element> {
     override func trim() {
         
     }
-}
+}
diff --git a/Sources/AllTestz/Anomalies.swift b/Sources/AllTestz/Anomalies.swift
@@ -0,0 +1 @@
+../../Tests/RxSwiftTests/Anomalies.swift
diff --git a/Sources/AllTestz/main.swift b/Sources/AllTestz/main.swift
diff --git a/Tests/RxSwiftTests/Anomalies.swift b/Tests/RxSwiftTests/Anomalies.swift
diff --git a/Tests/RxSwiftTests/Tests/SubjectConcurrencyTest.swift b/Tests/RxSwiftTests/Tests/SubjectConcurrencyTest.swift

Original file line number	Diff line number	Diff line change
`@@ -77,13 +77,13 @@ final class ShareReplay1<Element>`
`77`	`77`	`}`
`78`	`78`
`79`	`79`	`func on(event: Event<E>) {`
`80`		`- _lock.lock(); defer { _lock.unlock() }`
`81`		`- _synchronized_on(event)`
	`80`	`+ _synchronized_on(event).on(event)`
`82`	`81`	`}`
`83`	`82`
`84`		`- func _synchronized_on(event: Event<E>) {`
	`83`	`+ func _synchronized_on(_ event: Event<E>) -> Bag<AnyObserver<Element>> {`
	`84`	`+ _lock.lock(); defer { _lock.unlock() }`
`85`	`85`	`if _stopped {`
`86`		`- return`
	`86`	`+ return Bag()`
`87`	`87`	`}`
`88`	`88`
`89`	`89`	`switch event {`
`@@ -96,6 +96,6 @@ final class ShareReplay1<Element>`
`96`	`96`	`_connection = nil`
`97`	`97`	`}`
`98`	`98`
`99`		`- _observers.on(event)`
	`99`	`+ return _observers`
`100`	`100`	`}`
`101`		`-}`
	`101`	`+}`
Original file line number	Diff line number	Diff line change
`@@ -81,13 +81,13 @@ public final class BehaviorSubject<Element>`
`81`	`81`	`- parameter event: Event to send to the observers.`
`82`	`82`	`*/`
`83`	`83`	`public func on(event: Event<E>) {`
`84`		`- _lock.lock(); defer { _lock.unlock() }`
`85`		`- _synchronized_on(event)`
	`84`	`+ _synchronized_on(event).on(event)`
`86`	`85`	`}`
`87`	`86`
`88`		`- func _synchronized_on(event: Event<E>) {`
	`87`	`+ func _synchronized_on(event: Event<E>) -> Bag<AnyObserver<Element>> {`
	`88`	`+ _lock.lock(); defer { _lock.unlock() }`
`89`	`89`	`if _stoppedEvent != nil \|\| _disposed {`
`90`		`- return`
	`90`	`+ return Bag()`
`91`	`91`	`}`
`92`	`92`
`93`	`93`	`switch event {`
`@@ -97,7 +97,7 @@ public final class BehaviorSubject<Element>`
`97`	`97`	`_stoppedEvent = event`
`98`	`98`	`}`
`99`	`99`
`100`		`- _observers.on(event)`
	`100`	`+ return _observers`
`101`	`101`	`}`
`102`	`102`
`103`	`103`	`/**`
`@@ -158,4 +158,4 @@ public final class BehaviorSubject<Element>`
`158`	`158`	`_stoppedEvent = nil`
`159`	`159`	`}`
`160`	`160`	`}`
`161`		`-}`
	`161`	`+}`
Original file line number	Diff line number	Diff line change
`@@ -105,29 +105,30 @@ class ReplayBufferBase<Element>`
`105`	`105`	`}`
`106`	`106`
`107`	`107`	`override func on(event: Event<Element>) {`
`108`		`- _lock.lock(); defer { _lock.unlock() }`
`109`		`- _synchronized_on(event)`
	`108`	`+ _synchronized_on(event).on(event)`
`110`	`109`	`}`
`111`	`110`
`112`		`- func _synchronized_on(event: Event<E>) {`
	`111`	`+ func _synchronized_on(event: Event<E>) -> Bag<AnyObserver<Element>> {`
	`112`	`+ _lock.lock(); defer { _lock.unlock() }`
`113`	`113`	`if _disposed {`
`114`		`- return`
	`114`	`+ return Bag()`
`115`	`115`	`}`
`116`	`116`
`117`	`117`	`if _stoppedEvent != nil {`
`118`		`- return`
	`118`	`+ return Bag()`
`119`	`119`	`}`
`120`	`120`
`121`	`121`	`switch event {`
`122`	`122`	`case .Next(let value):`
`123`	`123`	`addValueToBuffer(value)`
`124`	`124`	`trim()`
`125`		`- _observers.on(event)`
	`125`	`+ return _observers`
`126`	`126`	`case .Error, .Completed:`
`127`	`127`	`_stoppedEvent = event`
`128`	`128`	`trim()`
`129`		`- _observers.on(event)`
	`129`	`+ let observers = _observers`
`130`	`130`	`_observers.removeAll()`
	`131`	`+ return observers`
`131`	`132`	`}`
`132`	`133`	`}`
`133`	`134`
`@@ -260,4 +261,4 @@ final class ReplayAll<Element> : ReplayManyBase<Element> {`
`260`	`261`	`override func trim() {`
`261`	`262`
`262`	`263`	`}`
`263`		`-}`
	`264`	`+}`
Original file line number	Diff line number	Diff line change
`@@ -0,0 +1 @@`
	`1`	`+../../Tests/RxSwiftTests/Anomalies.swift`