Improve logic in AtomicObserver to ensure synchronous disposing.

ReactiveKit.podspec

@@ -1,12 +1,12 @@
 Pod::Spec.new do |s|
   s.name             = "ReactiveKit"
-  s.version          = "3.5.4"
+  s.version          = "3.5.5"
   s.summary          = "A Swift Reactive Programming Framework"
   s.description      = "ReactiveKit is a Swift framework for reactive and functional reactive programming."
   s.homepage         = "https://github.com/ReactiveKit/ReactiveKit"
   s.license          = 'MIT'
   s.author           = { "Srdan Rasic" => "[email protected]" }
-  s.source           = { :git => "https://github.com/ReactiveKit/ReactiveKit.git", :tag => "v3.5.4" }
+  s.source           = { :git => "https://github.com/ReactiveKit/ReactiveKit.git", :tag => "v3.5.5" }
 
   s.ios.deployment_target       = '8.0'
   s.osx.deployment_target       = '10.9'

ReactiveKit.xcodeproj/xcshareddata/xcbaselines/ECBCCDD91BEB6B9B00723476.xcbaseline/FB4F14DE-1778-47BF-9AF1-A000D430FBBA.plist

@@ -0,0 +1,22 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
+<plist version="1.0">
+<dict>
+	<key>classNames</key>
+	<dict>
+		<key>SignalTests</key>
+		<dict>
+			<key>testPerformance()</key>
+			<dict>
+				<key>com.apple.XCTPerformanceMetric_WallClockTime</key>
+				<dict>
+					<key>baselineAverage</key>
+					<real>0.078778</real>
+					<key>baselineIntegrationDisplayName</key>
+					<string>Local Baseline</string>
+				</dict>
+			</dict>
+		</dict>
+	</dict>
+</dict>
+</plist>

ReactiveKit.xcodeproj/xcshareddata/xcbaselines/ECBCCDD91BEB6B9B00723476.xcbaseline/Info.plist

@@ -35,6 +35,37 @@
 				<string>com.apple.platform.iphonesimulator</string>
 			</dict>
 		</dict>
+		<key>FB4F14DE-1778-47BF-9AF1-A000D430FBBA</key>
+		<dict>
+			<key>localComputer</key>
+			<dict>
+				<key>busSpeedInMHz</key>
+				<integer>100</integer>
+				<key>cpuCount</key>
+				<integer>1</integer>
+				<key>cpuKind</key>
+				<string>Intel Core i5</string>
+				<key>cpuSpeedInMHz</key>
+				<integer>2700</integer>
+				<key>logicalCPUCoresPerPackage</key>
+				<integer>4</integer>
+				<key>modelCode</key>
+				<string>MacBookPro12,1</string>
+				<key>physicalCPUCoresPerPackage</key>
+				<integer>2</integer>
+				<key>platformIdentifier</key>
+				<string>com.apple.platform.macosx</string>
+			</dict>
+			<key>targetArchitecture</key>
+			<string>x86_64</string>
+			<key>targetDevice</key>
+			<dict>
+				<key>modelCode</key>
+				<string>iPhone9,1</string>
+				<key>platformIdentifier</key>
+				<string>com.apple.platform.iphonesimulator</string>
+			</dict>
+		</dict>
 	</dict>
 </dict>
 </plist>

ReactiveKit/Info.plist

@@ -15,7 +15,7 @@
 	<key>CFBundlePackageType</key>
 	<string>FMWK</string>
 	<key>CFBundleShortVersionString</key>
-	<string>3.5.4</string>
+	<string>3.5.5</string>
 	<key>CFBundleSignature</key>
 	<string>????</string>
 	<key>CFBundleVersion</key>

Sources/Observer.swift

@@ -60,27 +60,30 @@ public struct AnyObserver<Element, Error: Swift.Error>: ObserverProtocol {
 /// Observer that ensures events are sent atomically.
 public class AtomicObserver<Element, Error: Swift.Error>: ObserverProtocol {
 
-  private let observer: Observer<Element, Error>
-  private let disposable: Disposable
+  private var observer: Observer<Element, Error>?
   private let lock = NSRecursiveLock(name: "com.reactivekit.signal.atomicobserver")
-  private var terminated = false
+  private let parentDisposable: Disposable
+
+  public private(set) var disposable: Disposable!
 
   /// Creates an observer that wraps given closure.
   public init(disposable: Disposable, observer: @escaping Observer<Element, Error>) {
-    self.disposable = disposable
     self.observer = observer
+    self.parentDisposable = disposable
+    self.disposable = BlockDisposable { [weak self] in
+      self?.observer = nil
+      disposable.dispose()
+    }
   }
 
   /// Calles wrapped closure with the given element.
   public func on(_ event: Event<Element, Error>) {
     lock.lock(); defer { lock.unlock() }
-    guard !disposable.isDisposed && !terminated else { return }
-    if event.isTerminal {
-      terminated = true
-      observer(event)
-      disposable.dispose()
-    } else {
+    if let observer = observer {
       observer(event)
+      if event.isTerminal {
+        disposable.dispose()
+      }
     }
   }
 }

Sources/Signal.swift

@@ -39,6 +39,6 @@ public struct Signal<Element, Error: Swift.Error>: SignalProtocol {
     let serialDisposable = SerialDisposable(otherDisposable: nil)
     let observer = AtomicObserver(disposable: serialDisposable, observer: observer)
     serialDisposable.otherDisposable = producer(observer)
-    return serialDisposable
+    return observer.disposable
   }
 }