RxSwift combine observable with conditional

Question

I'm trying to combine observables and I want them to run in sequence (e.g., perform step 1, if some condition is met then perform step 2, if some condition is met then perform step 3). The only way I've found to do this is to add the conditions to each step, which I'm not a fan of: Here's a sample of my current solution:

enum Status {
    case unknown, exists, missing
}

func refresh() -> Observable<Status> {
    return checkLocalStatus()
        .flatMapLatest { $0 == .exists ? Observable.just($0) : self.attemptRemoteStatusOverride() }
        .flatMapLatest { $0 == .exists ? Observable.just($0) : self.attemptRemoteStatusUpdate() }
}

private func checkLocalStatus() -> Observable<Status> {
    return Observable.create { observer in
        // Regarding Maxim Volgin's comment, here I'm converting a closure to an 
        // observable... why not use Observable.create?
        self.cache.status { (status) in
            guard status != .exists else {
                observer.onNext(status) // .exists
                observer.onCompleted()
            }

            /* I don't want this condition to be here */
            if ignoreRemote {
                // status is !exists and we should ignore remote, throw error
                observer.onError(Errors.remoteDisabled)
            }

            observer.onNext(.missing)
            observer.onCompleted()
        }
    }
}

private func attemptRemoteStatusOverride() -> Observable<Status> {
    return remote.statusOverride()
}

private func attemptRemoteStatusUpdate() -> Observable<Status> {
    return Observable.create { observer in
        // Regarding Maxim Volgin's comment, here I'm converting a closure to an 
        // observable... why not use Observable.create?
        self.remote.updateStatus { (status, error) in
            guard error == nil else {
                observer.onError(error!)
            }
            observer.onNext(status)
            observer.onCompleted()
        }
    }
}

I'd like to do something like:

func refresh() -> Observable<Status> {
    return checkLocalStatus()
        .if({ $0 != .exists && !ignoreRemote },
            then: { self.attemptRemoteStatusOverride() },
            else: { return $0 })
        .if({ $0 != .exists },
            then: { self.attemptRemoteStatusUpdate() },
            else: { return $0 })
}

or

func refresh() -> Observable<Status> {
    return checkLocalStatus()
        .flatMapLatest(if: { $0 != .exists && !ignoreRemote }) { self.attemptRemoteStatusOverride() }
        .flatMapLatest(if: { $0 != .exists }) { self.attemptRemoteStatusUpdate() }
}

I haven't been able to find anything like what I'm attempting, so I assume I'm going about this wrong. Does anyone have suggestions or alternatives on how to go about this route of combining observables? I've seen examples using combineLatest and returning some results based on the result of something else, but I want to perform each step only if a condition is met. combineLatest would perform each step (every time) and then I would return the result(s) of some steps based on the output of other steps. I also started looking into writing a custom operator, but can't figure a way to do it.

Update: I've changed to the following and plan to write a method to remove duplication:

func refresh() -> Observable<Status> {
    return checkLocalStatus()
        .flatMapLatest { status -> Observable<Status>
            guard status != .exists && !ignoreRemote else {
                return Observable.just(status)
            }
            return self.attemptRemoteStatusOverride()
        }
        .flatMapLatest { status -> Observable<Status>
            guard status != .exists && !ignoreRemote else {
                return Observable.just(status)
            }
            return self.attemptRemoteStatusUpdate()
        }
}

Answer 1

Maybe you need some version of flatMapLatest function with conditions? You can make some function that does what you want with the syntax you want:

extension Observable {

    func flatMapLatest(condition: @escaping (E) -> Bool, then: @escaping (E) -> Observable, otherwise: @escaping () -> Observable) -> Observable {
        let observable = self.shareReplayLatestWhileConnected()
        let observableCondition = observable.map({ condition($0) }).shareReplayLatestWhileConnected()
        let observableThen: Observable<E> = observableCondition
            .filter({ $0 })
            .withLatestFrom(observable)
            .flatMapLatest({ then($0) })
            .shareReplayLatestWhileConnected()
        let observableOtherwise: Observable<E> = observableCondition
            .filter({ !$0 })
            .withLatestFrom(observable)
            .flatMapLatest({ _ in otherwise() })
            .shareReplayLatestWhileConnected()
        return Observable<Observable<E>>
            .from([observableThen, observableOtherwise])
            .merge()
    }
}

and use it

func refresh() -> Observable<Status> {
    let condition = { (status: Status) -> Bool in
        return status == .exists
    }
    let then = { (status: Status) -> Observable<Status> in
        return Observable.just(status)
    }
    return checkLocalStatus()
        .flatMapLatest(condition: condition, then: then, otherwise: self.attemptRemoteStatusOverride)
        .flatMapLatest(condition: condition, then: then, otherwise: self.attemptRemoteStatusUpdate)
}