Pretty significant refactor and improved memory/resource management.

Author: cloutiertyler

Sources/ColdSignal.swift

@@ -8,23 +8,21 @@
 
 import Foundation
 
-public final class ColdSignal<T, E: Error>: ColdSignalType, InternalSignalType, SpecialSignalGenerator {
+public final class ColdSignal<V, E: Swift.Error>: ColdSignalType, InternalSignalType, SpecialSignalGenerator {
 
-    public typealias Value = T
-    public typealias ErrorType = E
+    public typealias Value = V
+    public typealias Error = E
 
-    internal var observers = Bag<Observer<Value, ErrorType>>()
+    internal var observers = Bag<Observer<Value, Error>>()
 
     public var coldSignal: ColdSignal {
         return self
     }
 
-    internal let startHandler: (Observer<Value, ErrorType>) -> Disposable?
+    internal let startHandler: (Observer<Value, Error>) -> Disposable?
 
     private var cancelDisposable: Disposable?
 
-    private var handlerDisposable: Disposable?
-    
     private var started = false
 
     /// Initializes a ColdSignal that will invoke the given closure at the
@@ -38,7 +36,7 @@ public final class ColdSignal<T, E: Error>: ColdSignalType, InternalSignalType,
     /// 
     /// Invoking `start()` will have no effect until the signal is stopped. After
     /// `stop()` is called this process may be repeated.
-    public init(_ generator: @escaping (Observer<Value, ErrorType>) -> Disposable?) {
+    public init(_ generator: @escaping (Observer<Value, Error>) -> Disposable?) {
         self.startHandler = generator
     }
 
@@ -47,31 +45,19 @@ public final class ColdSignal<T, E: Error>: ColdSignalType, InternalSignalType,
     ///
     /// Returns a Disposable which can be used to interrupt the work associated
     /// with the signal and immediately send an `Interrupted` event.
-    
     @discardableResult
     public func start() {
         if !started {
             started = true
 
-            let observer = Observer<Value, ErrorType> { event in
-                // Pass event downstream
-                self.observers.forEach { (observer) in
-                    observer.action(event)
-                }
-                
-                // If event is terminating dispose of the handlerDisposable.
-                if event.isTerminating {
-                    self.handlerDisposable?.dispose()
-                }
-            }
-            
-            handlerDisposable = startHandler(observer)
+            let observer = Observer(with: CircuitBreaker(holding: self))
+            let handlerDisposable = startHandler(observer)
 
             // The cancel disposable should send interrupted and then dispose of the 
             // disposable produced by the startHandler.
-            cancelDisposable = ActionDisposable { [weak self] in
+            cancelDisposable = ActionDisposable {
                 observer.sendInterrupted()
-                self?.handlerDisposable?.dispose()
+                handlerDisposable?.dispose()
             }
         }
     }
@@ -95,7 +81,7 @@ extension ColdSignal: CustomDebugStringConvertible {
 public protocol ColdSignalType: SignalType {
 
     /// The exposed raw signal that underlies the ColdSignalType
-    var coldSignal: ColdSignal<Value, ErrorType> { get }
+    var coldSignal: ColdSignal<Value, Error> { get }
 
     /// Invokes the closure provided upon initialization, and passes in a newly
     /// created observer to which events can be sent.
@@ -110,7 +96,7 @@ public protocol ColdSignalType: SignalType {
 
 public extension ColdSignalType {
 
-    public var signal: Signal<Value, ErrorType> {
+    public var signal: Signal<Value, Error> {
         return Signal { observer in
             self.coldSignal.add(observer: observer)
         }
@@ -140,7 +126,7 @@ public extension ColdSignalType {
     /// Returns a Disposable which can be used to disconnect the observer. Disposing
     /// of the Disposable will have no effect on the Signal itself.
     @discardableResult
-    public func add(observer: Observer<Value, ErrorType>) -> Disposable? {
+    public func add(observer: Observer<Value, Error>) -> Disposable? {
         let token = coldSignal.observers.insert(value: observer)
         return ActionDisposable {
             self.coldSignal.observers.removeValueForToken(token: token)
@@ -152,7 +138,7 @@ public extension ColdSignalType {
     ///
     /// Returns a Disposable which can be used to dispose of the added observer.
     @discardableResult
-    public func start(with observer: Observer<Value, ErrorType>) -> Disposable? {
+    public func start(with observer: Observer<Value, Error>) -> Disposable? {
         let disposable = coldSignal.add(observer: observer)
         self.coldSignal.start()
         return disposable
@@ -163,7 +149,7 @@ public extension ColdSignalType {
     ///
     /// Returns a Disposable which can be used to dispose of the added observer.
     @discardableResult
-    public func start(_ observerAction: @escaping Observer<Value, ErrorType>.Action) -> Disposable? {
+    public func start(_ observerAction: @escaping Observer<Value, Error>.Action) -> Disposable? {
         return start(with: Observer(observerAction))
     }
 
@@ -190,7 +176,7 @@ public extension ColdSignalType {
     ///
     /// Returns a Disposable which can be used to dispose of the added observer.
     @discardableResult
-    public func startWithFailed(failed: @escaping (ErrorType) -> Void) -> Disposable? {
+    public func startWithFailed(failed: @escaping (Error) -> Void) -> Disposable? {
         return start(with: Observer(failed: failed))
     }
 
@@ -212,18 +198,18 @@ public extension ColdSignalType {
     ///
     /// The new `ColdSignal` is in no way related to the source `ColdSignal` except
     /// that they share a reference to the same `startHandler`.
-    public func lift<U, F>(_ transform: @escaping (Signal<Value, ErrorType>) -> Signal<U, F>) -> ColdSignal<U, F> {
+    public func lift<U, F>(_ transform: @escaping (Signal<Value, Error>) -> Signal<U, F>) -> ColdSignal<U, F> {
         return ColdSignal { observer in
-            let (pipeSignal, pipeObserver) = Signal<Value, ErrorType>.pipe()
+            let (pipeSignal, pipeObserver) = Signal<Value, Error>.pipe()
             transform(pipeSignal).add(observer: observer)
             return self.coldSignal.startHandler(pipeObserver)
         }
     }
 
-    public func lift<U, F>(_ transform: @escaping (Signal<Value, ErrorType>) -> (Signal<U, F>, Signal<U, F>))
+    public func lift<U, F>(_ transform: @escaping (Signal<Value, Error>) -> (Signal<U, F>, Signal<U, F>))
         -> (ColdSignal<U, F>, ColdSignal<U, F>)
     {
-        let (pipeSignal, pipeObserver) = Signal<Value, ErrorType>.pipe()
+        let (pipeSignal, pipeObserver) = Signal<Value, Error>.pipe()
         let (left, right) = transform(pipeSignal)
         let coldLeft = ColdSignal<U, F> { observer in
             left.add(observer: observer)
@@ -237,33 +223,33 @@ public extension ColdSignalType {
     }
 
     /// Maps each value in the signal to a new value.
-    public func map<U>(_ transform: @escaping (Value) -> U) -> ColdSignal<U, ErrorType> {
+    public func map<U>(_ transform: @escaping (Value) -> U) -> ColdSignal<U, Error> {
         return lift { $0.map(transform) }
     }
 
     /// Maps errors in the signal to a new error.
-    public func mapError<F>(_ transform: @escaping (ErrorType) -> F) -> ColdSignal<Value, F> {
+    public func mapError<F>(_ transform: @escaping (Error) -> F) -> ColdSignal<Value, F> {
         return lift { $0.mapError(transform) }
     }
 
     /// Preserves only the values of the signal that pass the given predicate.
-    public func filter(_ predicate: @escaping (Value) -> Bool) -> ColdSignal<Value, ErrorType> {
+    public func filter(_ predicate: @escaping (Value) -> Bool) -> ColdSignal<Value, Error> {
         return lift { $0.filter(predicate) }
     }
 
     /// Splits the signal into two signals. The first signal in the tuple matches the
     /// predicate, the second signal does not match the predicate
     public func partition(_ predicate: @escaping (Value) -> Bool)
-        -> (ColdSignal<Value, ErrorType>, ColdSignal<Value, ErrorType>) {
+        -> (ColdSignal<Value, Error>, ColdSignal<Value, Error>) {
         return lift { $0.partition(predicate) }
     }
 
     /// Aggregate values into a single combined value. Mirrors the Swift Collection
-    public func reduce<T>(initial: T, _ combine: @escaping (T, Value) -> T) -> ColdSignal<T, ErrorType> {
+    public func reduce<T>(initial: T, _ combine: @escaping (T, Value) -> T) -> ColdSignal<T, Error> {
         return lift { $0.reduce(initial: initial, combine) }
     }
 
-    public func flatMap<U>(_ transform: @escaping (Value) -> U?) -> ColdSignal<U, ErrorType> {
+    public func flatMap<U>(_ transform: @escaping (Value) -> U?) -> ColdSignal<U, Error> {
         return lift { $0.flatMap(transform) }
     }
 

Sources/Event.swift

@@ -11,93 +11,93 @@ import Foundation
 /// Represents a signal event.
 ///
 /// Signals must conform to the grammar:
-/// `Next* (Failed | Completed | Interrupted)?`
-public enum Event<Value, ErrorType: Error> {
+/// `next* (failed | completed | interrupted)?`
+public enum Event<Value, Error: Swift.Error> {
 
     /// A value provided by the signal.
-    case Next(Value)
+    case next(Value)
 
     /// The signal terminated because of an error. No further events will be
     /// received.
-    case Failed(ErrorType)
+    case failed(Error)
 
     /// The signal successfully terminated. No further events will be received.
-    case Completed
+    case completed
 
     /// Event production on the signal has been interrupted. No further events
     /// will be received.
-    case Interrupted
+    case interrupted
 
 
     /// Whether this event indicates signal termination (i.e., that no further
     /// events will be received).
     public var isTerminating: Bool {
         switch self {
-        case .Next:
+        case .next:
             return false
 
-        case .Failed, .Completed, .Interrupted:
+        case .failed, .completed, .interrupted:
             return true
         }
     }
 
     /// Lifts the given function over the event's value.
-    public func map<U>(_ f: (Value) -> U) -> Event<U, ErrorType> {
+    public func map<U>(_ f: (Value) -> U) -> Event<U, Error> {
         switch self {
-        case let .Next(value):
-            return .Next(f(value))
+        case let .next(value):
+            return .next(f(value))
 
-        case let .Failed(error):
-            return .Failed(error)
+        case let .failed(error):
+            return .failed(error)
 
-        case .Completed:
-            return .Completed
+        case .completed:
+            return .completed
 
-        case .Interrupted:
-            return .Interrupted
+        case .interrupted:
+            return .interrupted
         }
     }
 
     /// Lifts the given function over the event's value.
-    public func flatMap<U>(_ f: (Value) -> U?) -> Event<U, ErrorType>? {
+    public func flatMap<U>(_ f: (Value) -> U?) -> Event<U, Error>? {
         switch self {
-        case let .Next(value):
+        case let .next(value):
             if let nextValue = f(value) {
-                return .Next(nextValue)
+                return .next(nextValue)
             }
             return nil
 
-        case let .Failed(error):
-            return .Failed(error)
+        case let .failed(error):
+            return .failed(error)
 
-        case .Completed:
-            return .Completed
+        case .completed:
+            return .completed
 
-        case .Interrupted:
-            return .Interrupted
+        case .interrupted:
+            return .interrupted
         }
     }
 
     /// Lifts the given function over the event's error.
-    public func mapError<F>(_ f: (ErrorType) -> F) -> Event<Value, F> {
+    public func mapError<F>(_ f: (Error) -> F) -> Event<Value, F> {
         switch self {
-        case let .Next(value):
-            return .Next(value)
+        case let .next(value):
+            return .next(value)
 
-        case let .Failed(error):
-            return .Failed(f(error))
+        case let .failed(error):
+            return .failed(f(error))
 
-        case .Completed:
-            return .Completed
+        case .completed:
+            return .completed
 
-        case .Interrupted:
-            return .Interrupted
+        case .interrupted:
+            return .interrupted
         }
     }
 
     /// Unwraps the contained `Next` value.
     public var value: Value? {
-        if case let .Next(value) = self {
+        if case let .next(value) = self {
             return value
         } else {
             return nil
@@ -106,26 +106,26 @@ public enum Event<Value, ErrorType: Error> {
 
     /// Unwraps the contained `Error` value.
     public var error: Error? {
-        if case let .Failed(error) = self {
+        if case let .failed(error) = self {
             return error
         } else {
             return nil
         }
     }
 }
 
-public func == <Value: Equatable, ErrorType: Equatable> (lhs: Event<Value, ErrorType>, rhs: Event<Value, ErrorType>) -> Bool {
+public func == <Value: Equatable, Error: Equatable> (lhs: Event<Value, Error>, rhs: Event<Value, Error>) -> Bool {
     switch (lhs, rhs) {
-    case let (.Next(left), .Next(right)):
+    case let (.next(left), .next(right)):
         return left == right
 
-    case let (.Failed(left), .Failed(right)):
+    case let (.failed(left), .failed(right)):
         return left == right
 
-    case (.Completed, .Completed):
+    case (.completed, .completed):
         return true
 
-    case (.Interrupted, .Interrupted):
+    case (.interrupted, .interrupted):
         return true
 
     default: