Refactor internal implementation and add more test cases

Author: ishkawa

Action.swift

@@ -23,157 +23,114 @@ public final class Action<Input, Element> {
     /// All inputs are always appear in this subject even if the action is not enabled.
     /// Thus, inputs count equals elements count + errors count.
     public let inputs = PublishSubject<Input>()
-    fileprivate let _completed = PublishSubject<Void>()
 
-    /// Errors aggrevated from invocations of execute(). 
+    /// Errors aggrevated from invocations of execute().
     /// Delivered on whatever scheduler they were sent from.
-    public var errors: Observable<ActionError> {
-        return self._errors.asObservable()
-    }
-    fileprivate let _errors = PublishSubject<ActionError>()
+    public let errors: Observable<ActionError>
 
-    /// Whether or not we're currently executing. 
+    /// Whether or not we're currently executing.
     /// Delivered on whatever scheduler they were sent from.
-    public var elements: Observable<Element> {
-        return self._elements.asObservable()
-    }
-    fileprivate let _elements = PublishSubject<Element>()
+    public let elements: Observable<Element>
 
     /// Whether or not we're currently executing. 
     /// Always observed on MainScheduler.
-    public var executing: Observable<Bool> {
-        return self._executing.asObservable().observeOn(MainScheduler.instance)
-    }
-    fileprivate let _executing = Variable(false)
-    
+    public let executing: Observable<Bool>
+
     /// Observables returned by the workFactory.
     /// Useful for sending results back from work being completed
     /// e.g. response from a network call.
-    public var executionObservables: Observable<Observable<Element>> {
-        return self._executionObservables.asObservable().observeOn(MainScheduler.instance)
-    }
-    fileprivate let _executionObservables = PublishSubject<Observable<Element>>()
-    
+    public let executionObservables: Observable<Observable<Element>>
+
     /// Whether or not we're enabled. Note that this is a *computed* sequence
     /// property based on enabledIf initializer and if we're currently executing.
     /// Always observed on MainScheduler.
-    public var enabled: Observable<Bool> {
-        return _enabled.asObservable().observeOn(MainScheduler.instance)
-    }
-    public fileprivate(set) var _enabled = BehaviorSubject(value: true)
+    public let enabled: Observable<Bool>
 
-    fileprivate let executingQueue = DispatchQueue(label: "com.ashfurrow.Action.executingQueue", attributes: [])
-    fileprivate let disposeBag = DisposeBag()
+    private let disposeBag = DisposeBag()
 
-    public init(enabledIf: Observable<Bool> = Observable.just(true), workFactory: @escaping WorkFactory) {
-        self._enabledIf = enabledIf
+    public init(
+        enabledIf: Observable<Bool> = Observable.just(true),
+        workFactory: @escaping WorkFactory) {
 
+        self._enabledIf = enabledIf
         self.workFactory = workFactory
 
-        Observable.combineLatest(self._enabledIf, self.executing) { (enabled, executing) -> Bool in
-            return enabled && !executing
-        }.bindTo(_enabled).addDisposableTo(disposeBag)
-
-        self.inputs.subscribe(onNext: { [weak self] (input) in
-            self?._execute(input)
-        }).addDisposableTo(disposeBag)
-    }
-}
-
-
-// MARK: Execution!
-public extension Action {
-
-    @discardableResult
-    public func execute(_ input: Input) -> Observable<Element> {
-        let buffer = ReplaySubject<Element>.createUnbounded()
-        let error = errors
-            .flatMap { error -> Observable<Element> in
-                if case .underlyingError(let error) = error {
-                    throw error
+        let enabledSubject = BehaviorSubject<Bool>(value: false)
+        enabled = enabledSubject.asObservable()
+        
+        executionObservables = inputs
+            .withLatestFrom(enabled) { $0 }
+            .flatMap { input, enabled -> Observable<Observable<Element>> in
+                if enabled {
+                    return Observable.of(workFactory(input).shareReplay(1))
                 } else {
                     return Observable.empty()
                 }
             }
+            .shareReplay(1)
+
+        elements = executionObservables
+            .flatMap { $0.catchError { _ in Observable.empty() } }
+
+        let notEnabledError = inputs
+            .withLatestFrom(enabled)
+            .flatMap { $0 ? Observable.empty() : Observable.of(ActionError.notEnabled) }
+
+        let underlyingError = executionObservables
+            .flatMap { elements in
+                return elements
+                    .flatMap { _ in Observable<ActionError>.never() }
+                    .catchError { error in
+                        if let actionError = error as? ActionError {
+                            return Observable.of(actionError)
+                        } else {
+                            return Observable.of(.underlyingError(error))
+                        }
+                    }
+            }
 
-        Observable
-            .of(elements, error)
+        errors = Observable
+            .of(notEnabledError, underlyingError)
             .merge()
-            .takeUntil(_completed)
-            .bindTo(buffer)
-            .addDisposableTo(disposeBag)
-        
-        inputs.onNext(input)
-        
-        return buffer.asObservable()
-    }
-
-    @discardableResult
-    fileprivate func _execute(_ input: Input) -> Observable<Element> {
-
-        // Buffer from the work to a replay subject.
-        let buffer = ReplaySubject<Element>.createUnbounded()
 
-        // See if we're already executing.
-        var startedExecuting = false
-        self.doLocked {
-            if self._enabled.valueOrFalse {
-                self._executing.value = true
-                startedExecuting = true
+        let executionStart = executionObservables.catchError { _ in Observable.empty() }
+        let executionEnd = executionObservables
+            .catchError { _ in Observable.empty() }
+            .flatMap { observable -> Observable<Void> in
+                return observable
+                    .flatMap { _ in Observable<Void>.empty() }
+                    .concat(Observable.just())
+                    .catchErrorJustReturn()
             }
-        }
-
-        // Make sure we started executing and we're accidentally disabled.
-        guard startedExecuting else {
-            let error = ActionError.notEnabled
-            self._errors.onNext(error)
-            buffer.onError(error)
-
-            return buffer
-        }
-
-        let work = self.workFactory(input)
-        defer {
-            // Subscribe to the work.
-            work.multicast(buffer).connect().addDisposableTo(disposeBag)
-        }
-
-		self._executionObservables.onNext(buffer)
-
-        buffer.subscribe(onNext: {[weak self] element in
-                    self?._elements.onNext(element)
-                },
-                onError: {[weak self] error in
-                    self?._errors.onNext(ActionError.underlyingError(error))
-                },
-                onCompleted: {[weak self] in
-                    self?._completed.onNext()
-                },
-                onDisposed: {[weak self] in
-                    self?.doLocked { self?._executing.value = false }
-                })
-            .addDisposableTo(disposeBag)
 
+        executing = Observable
+            .of(executionStart.map { _ in true }, executionEnd.map { _ in false })
+            .merge()
+            .startWith(false)
 
-        return buffer.asObservable()
+        Observable
+            .combineLatest(executing, enabledIf) { !$0 && $1 }
+            .bindTo(enabledSubject)
+            .addDisposableTo(disposeBag)
     }
-}
 
-private extension Action {
-    func doLocked(_ closure: () -> Void) {
-        executingQueue.sync(execute: closure)
-    }
-}
+    @discardableResult
+    public func execute(_ value: Input) -> Observable<Element> {
+        let subject = ReplaySubject<Element>.createUnbounded()
 
-fileprivate let executingQueue = DispatchQueue(label: "com.ashfurrow.Action.executingQueue", attributes: [])
-internal func doLocked(_ closure: () -> Void) {
-    executingQueue.sync(execute: closure)
-}
+        executionObservables
+            .take(1)
+            .flatMap { $0.catchError { _ in Observable.never() } }
+            .bindTo(subject)
+            .addDisposableTo(disposeBag)
+
+        errors
+            .map { throw $0 }
+            .bindTo(subject)
+            .addDisposableTo(disposeBag)
 
-internal extension BehaviorSubject where Element: ExpressibleByBooleanLiteral {
-    var valueOrFalse: Element {
-        guard let value = try? value() else { return false }
+        inputs.onNext(value)
 
-        return value
+        return subject.asObservable()
     }
 }

AlertAction.swift

@@ -20,27 +20,23 @@ public extension Reactive where Base: UIAlertAction {
     public var action: CocoaAction? {
         get {
             var action: CocoaAction?
-            doLocked {
-                action = objc_getAssociatedObject(base, &AssociatedKeys.Action) as? Action
-            }
+            action = objc_getAssociatedObject(base, &AssociatedKeys.Action) as? Action
             return action
         }
 
         set {
-            doLocked {
-                // Store new value.
-                objc_setAssociatedObject(base, &AssociatedKeys.Action, newValue, .OBJC_ASSOCIATION_RETAIN_NONATOMIC)
-
-                // This effectively disposes of any existing subscriptions.
-                self.base.resetActionDisposeBag()
-
-                // Set up new bindings, if applicable.
-                if let action = newValue {
-                    action
-                        .enabled
-                        .bindTo(self.enabled)
-                        .addDisposableTo(self.base.actionDisposeBag)
-                }
+            // Store new value.
+            objc_setAssociatedObject(base, &AssociatedKeys.Action, newValue, .OBJC_ASSOCIATION_RETAIN_NONATOMIC)
+            
+            // This effectively disposes of any existing subscriptions.
+            self.base.resetActionDisposeBag()
+            
+            // Set up new bindings, if applicable.
+            if let action = newValue {
+                action
+                    .enabled
+                    .bindTo(self.enabled)
+                    .addDisposableTo(self.base.actionDisposeBag)
             }
         }
     }