Merge pull request #73 from fpillet/rewrite-errors-observable

Rewrite errors observable

Author: ashfurrow

Sources/Action/Action.swift

@@ -61,13 +61,19 @@ public final class Action<Input, Element> {
 
         let enabledSubject = BehaviorSubject<Bool>(value: false)
         enabled = enabledSubject.asObservable()
+
+        let errorsSubject = PublishSubject<ActionError>()
+        errors = errorsSubject.asObservable()
 
         executionObservables = inputs
             .withLatestFrom(enabled) { $0 }
             .flatMap { input, enabled -> Observable<Observable<Element>> in
                 if enabled {
-                    return Observable.of(workFactory(input).shareReplay(1))
+                    return Observable.of(workFactory(input)
+                                             .do(onError: { errorsSubject.onNext(.underlyingError($0)) })
+                                             .shareReplay(1))
                 } else {
+                    errorsSubject.onNext(.notEnabled)
                     return Observable.empty()
                 }
             }
@@ -76,27 +82,6 @@ public final class Action<Input, Element> {
         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))
-                        }
-                    }
-            }
-        
-        errors = Observable
-            .of(notEnabledError, underlyingError)
-            .merge()
-
         executing = executionObservables.flatMap {
                 execution -> Observable<Bool> in
                 let execution = execution
@@ -122,23 +107,20 @@ public final class Action<Input, Element> {
             inputs.onNext(value)
         }
 
-        let execution = executionObservables
-            .take(1)
-            .flatMap { $0 }
-            .catchError { throw ActionError.underlyingError($0) }
-
-        let notEnabledError = inputs
-            .takeUntil(executionObservables)
-            .withLatestFrom(enabled)
-            .flatMap { $0 ? Observable<Element>.empty() : Observable.error(ActionError.notEnabled) }
-
-        let subject = ReplaySubject<Element>.createUnbounded()
-        Observable
-            .of(execution, notEnabledError)
-            .merge()
-            .subscribe(subject)
-            .addDisposableTo(disposeBag)
-
-        return subject.asObservable()
+		let subject = ReplaySubject<Element>.createUnbounded()
+		
+		let work = executionObservables
+			.map { $0.catchError { throw ActionError.underlyingError($0) } }
+		
+		let error = errors
+			.map { Observable<Element>.error($0) }
+
+		work.amb(error)
+			.take(1)
+			.flatMap { $0 }
+			.subscribe(subject)
+			.addDisposableTo(disposeBag)
+		
+		return subject.asObservable()
     }
 }