Cleanup old code.

RxSwift/RxSwift/Observables/Implementations/AsObservable.swift

@@ -8,17 +8,16 @@
 
 import Foundation
 
-class AsObservableSink_<ElementType> : ObserverType, Disposable {
+class AsObservableSink_<ElementType> : Sink<ElementType>, ObserverType {
     typealias Element = ElementType
     
-    let sink: Sink<Element>
- 
-    func dispose() {
-        sink.dispose()
+    override init(observer: ObserverOf<Element>, cancel: Disposable) {
+        super.init(observer: observer, cancel: cancel)
     }
     
     func on(event: Event<Element>) {
-        self.sink.state.observer.on(event)
+        self.observer.on(event)
+        
         switch event {
         case .Error: fallthrough
         case .Completed:
@@ -27,9 +26,6 @@ class AsObservableSink_<ElementType> : ObserverType, Disposable {
         }
     }
     
-    init(observer: ObserverOf<Element>, cancel: Disposable) {
-        self.sink = Sink(observer: observer, cancel: cancel)
-    }
 }
 
 class AsObservable<Element> : Producer<Element> {

RxSwift/RxSwift/Observables/Implementations/DistinctUntilChanged.swift

@@ -24,24 +24,25 @@ class DistinctUntilChanged_<ElementType, Key>: Sink<ElementType>, ObserverType {
         
         switch event {
         case .Next(let value):
-            let keyResult = self.parent.selector(value.value)
-                
-            let areEqualResult = keyResult >== { key -> Result<Bool> in
+            self.parent.selector(value.value) >== { key in
+                var areEqual: Result<Bool>
                 if let currentKey = self.currentKey {
-                    return self.parent.comparer(currentKey, key)
+                    areEqual = self.parent.comparer(currentKey, key)
                 }
                 else {
-                    return success(false)
+                    areEqual = success(false)
                 }
-            } >== { areEqual in
-                if areEqual {
+                
+                return areEqual >== { areEqual in
+                    if areEqual {
+                        return SuccessResult
+                    }
+                    
+                    self.currentKey = key
+                    
+                    observer.on(event)
                     return SuccessResult
                 }
-                
-                self.currentKey = *keyResult
-                
-                observer.on(event)
-                return SuccessResult
             } >>! { error -> Result<Void> in
                 observer.on(.Error(error))
                 self.dispose()

RxSwift/RxSwift/Observables/Implementations/Do.swift

@@ -22,11 +22,11 @@ class Do_<ElementType> : Sink<ElementType>, ObserverType, Disposable {
     func on(event: Event<Element>) {
         parent.eventHandler(event) >>! { error in
             // catch clause
-            self.state.observer.on(Event.Error(error))
+            self.observer.on(.Error(error))
             self.dispose()
             return SuccessResult
         } >== { _ -> Result<Void> in
-            self.state.observer.on(event)
+            self.observer.on(event)
             if event.isStopEvent {
                 self.dispose()
             }

RxTests/RxSwiftTests/Tests/Observable+StandardSequenceOperatorsTest.swift

@@ -36,7 +36,8 @@ func isPrime(i: Int) -> Bool {
     return true
 }
 
-// Where Tests
+// where
+
 extension ObservableStandardSequenceOperators  {
     func test_whereComplete() {
         let scheduler = TestScheduler(initialClock: 0)