import Foundation import RxSwift import RxCocoa /// Typealias for compatibility with UIButton's rx.action property. public typealias CocoaAction = Action /// Typealias for actions with work factory returns `Completable`. public typealias CompletableAction = Action /// Possible errors from invoking execute() public enum ActionError: Error { case notEnabled case underlyingError(Error) } /** Represents a value that accepts a workFactory which takes some Observable as its input and produces an Observable as its output. When this excuted via execute() or inputs subject, it passes its parameter to this closure and subscribes to the work. */ public final class Action { public typealias WorkFactory = (Input) -> Observable public let _enabledIf: Observable public let workFactory: WorkFactory /// Inputs that triggers execution of action. /// This subject also includes inputs as aguments of execute(). /// 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 = InputSubject() /// Errors aggrevated from invocations of execute(). /// Delivered on whatever scheduler they were sent from. public let errors: Observable /// Whether or not we're currently executing. /// Delivered on whatever scheduler they were sent from. public let elements: Observable /// Whether or not we're currently executing. public let executing: Observable /// Observables returned by the workFactory. /// Useful for sending results back from work being completed /// e.g. response from a network call. public let executionObservables: Observable> /// 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 let enabled: Observable private let disposeBag = DisposeBag() public convenience init( enabledIf: Observable = Observable.just(true), workFactory: @escaping (Input) -> O ) where O.Element == Element { self.init(enabledIf: enabledIf) { workFactory($0).asObservable() } } public init( enabledIf: Observable = Observable.just(true), workFactory: @escaping WorkFactory) { self._enabledIf = enabledIf self.workFactory = workFactory let enabledSubject = BehaviorSubject(value: false) enabled = enabledSubject.asObservable() let errorsSubject = PublishSubject() errors = errorsSubject.asObservable() executionObservables = inputs .withLatestFrom(enabled) { input, enabled in (input, enabled) } .flatMap { input, enabled -> Observable> in if enabled { return Observable.of(workFactory(input) .do(onError: { errorsSubject.onNext(.underlyingError($0)) }) .share(replay: 1, scope: .forever)) } else { errorsSubject.onNext(.notEnabled) return Observable.empty() } } .share() elements = executionObservables .flatMap { $0.catch { _ in Observable.empty() } } executing = executionObservables.flatMap { execution -> Observable in let execution = execution .flatMap { _ in Observable.empty() } .catch { _ in Observable.empty() } return Observable.concat([Observable.just(true), execution, Observable.just(false)]) } .startWith(false) .share(replay: 1, scope: .forever) Observable .combineLatest(executing, enabledIf) { !$0 && $1 } .bind(to: enabledSubject) .disposed(by: disposeBag) } @discardableResult public func execute(_ value: Input) -> Observable { defer { inputs.onNext(value) } let subject = ReplaySubject.createUnbounded() let work = executionObservables .map { $0.catch { throw ActionError.underlyingError($0) } } let error = errors .map { Observable.error($0) } work.amb(error) .take(1) .flatMap { $0 } .subscribe(subject) .disposed(by: disposeBag) return subject.asObservable() } }