Sink modifications.

RxSwift/Observables/Implementations/Sink.swift

@@ -9,18 +9,20 @@
 import Foundation
 
 class Sink<O : ObserverType> : SingleAssignmentDisposable {
-    private var _lock = SpinLock()
-
     // state
-    private var _observer: O?
+    private var _observer: O
 
     var observer: O? {
         get {
-            _lock.lock(); defer { _lock.unlock() }
+            //_lock.lock(); defer { _lock.unlock() }
             return _observer
         }
     }
-    
+
+    func forward(event: Event<O.E>) {
+        _observer.on(event)
+    }
+
     init(observer: O) {
 #if TRACE_RESOURCES
         OSAtomicIncrement32(&resourceCount)
@@ -28,19 +30,6 @@ class Sink<O : ObserverType> : SingleAssignmentDisposable {
         _observer = observer
     }
 
-    private func _disposeObserver() {
-        _lock.lock(); defer { _lock.unlock() }
-
-        _observer = nil
-    }
-    
-    override func dispose() {
-        if !disposed {
-            _disposeObserver()
-        }
-        super.dispose()
-    }
-    
     deinit {
 #if TRACE_RESOURCES
         OSAtomicDecrement32(&resourceCount)

RxTests/PerformanceTests/main.swift

@@ -24,14 +24,14 @@ compareTwoImplementations(benchmarkTime: true, first: {
     //let a = just(1)
 
     //combineLatest(a,
-        publishSubject
+        publishSubject//.asDriver(onErrorJustReturn: -1)
         .shareReplay(1)
         .map { $0 }
-        .filter { _ in true }// ){ x, _ in x }
-        //.map { $0 }
-        //.flatMap { just($0) }
+        .filter { _ in true }//){ x, _ in x }
+        .map { $0 }
+        .flatMap { just($0) }
         .subscribeNext { _ in
-            
+
         }