Fix error handling in execute()

Author: ishkawa

Sources/Action/Action.swift

@@ -120,20 +120,26 @@ public final class Action<Input, Element> {
 
     @discardableResult
     public func execute(_ value: Input) -> Observable<Element> {
-        let subject = ReplaySubject<Element>.createUnbounded()
-
-        executionObservables
+        defer {
+            inputs.onNext(value)
+        }
+        
+        let execution = executionObservables
             .take(1)
-            .flatMap { $0.catchError { _ in Observable.never() } }
-            .bindTo(subject)
-            .addDisposableTo(disposeBag)
+            .flatMap { $0 }
+            .catchError { throw ActionError.underlyingError($0) }
 
-        errors
-            .map { throw $0 }
-            .bindTo(subject)
-            .addDisposableTo(disposeBag)
+        let notEnabledError = inputs
+            .takeUntil(executionObservables)
+            .withLatestFrom(enabled)
+            .flatMap { $0 ? Observable<Element>.empty() : Observable.error(ActionError.notEnabled) }
 
-        inputs.onNext(value)
+        let subject = ReplaySubject<Element>.createUnbounded()
+        Observable
+            .of(execution, notEnabledError)
+            .merge()
+            .subscribe(subject)
+            .addDisposableTo(disposeBag)
 
         return subject.asObservable()
     }

Tests/ActionTests/ActionTests.swift

@@ -473,6 +473,56 @@ class ActionTests: QuickSpec {
                     expect(executionObservables.events).to(beEmpty())
                 }
             }
+
+            context("execute while executing") {
+                var secondElement: TestableObserver<String>!
+                var trigger: PublishSubject<Void>!
+                
+                beforeEach {
+                    secondElement = scheduler.createObserver(String.self)
+                    trigger = PublishSubject<Void>()
+                    action = Action { Observable.just($0).sample(trigger) }
+                    
+                    action.executionObservables
+                        .bindTo(executionObservables)
+                        .addDisposableTo(disposeBag)
+                    
+                    scheduler.scheduleAt(10) {
+                        action.execute("a")
+                            .bindTo(element)
+                            .addDisposableTo(disposeBag)
+                    }
+
+                    scheduler.scheduleAt(20) {
+                        action.execute("b")
+                            .bindTo(secondElement)
+                            .addDisposableTo(disposeBag)
+                    }
+
+                    scheduler.scheduleAt(30) {
+                        trigger.onNext()
+                    }
+
+                    scheduler.start()
+                }
+
+                it("first element receives single value") {
+                    XCTAssertEqual(element.events, [
+                        next(30, "a"),
+                        completed(30),
+                    ])
+                }
+
+                it("second element fails with notEnabled error") {
+                    XCTAssertEqual(secondElement.events, [
+                        error(20, ActionError.notEnabled)
+                    ])
+                }
+
+                it("executes once") {
+                    expect(executionObservables.events.count) == 1
+                }
+            }
         }
     }
 }