iOS RxSwift how to prevent sequence from being disposed on (throw error)?

Question

I have a sequence made up of multiple operators. There are total of 7 places where errors can be generated during this sequence processing. I'm running into an issue where the sequence does not behave as I expected and I'm looking for an elegant solution around the problem:

let inputRelay = PublishRelay<Int>()
let outputRelay = PublishRelay<Result<Int>>()

inputRelay
.map{ /*may throw multiple errors*/}
.flatmap{ /*may throw error*/ }
.map{}
.filter{}
.map{ _ -> Result<Int> in ...}
.catchError{}
.bind(to: outputRelay)

I thought that catchError would simply catch the error, allow me to convert it to failure result, but prevent the sequence from being deallocated. However, I see that the first time an error is caught, the entire sequence is deallocated and no more events go through.

Without this behavior, I'm left with a fugly Results<> all over the place, and have to branch my sequence multiple times to direct the Result.failure(Error) to the output. There are non-recoverable errors, so retry(n) is not an option:

let firstOp = inputRelay
.map{ /*may throw multiple errors*/}
.share()

//--Handle first error results--
firstOp
.filter{/*errorResults only*/}
.bind(to: outputRelay)

let secondOp = firstOp
.flatmap{ /*may throw error*/ }
.share()

//--Handle second error results--
secondOp
.filter{/*errorResults only*/}
.bind(to: outputRelay)

secondOp
.map{}
.filter{}
.map{ _ -> Result<Int> in ...}
.catchError{}
.bind(to: outputRelay)

^ Which is very bad, because there are around 7 places where errors can be thrown and I cannot just keep branching the sequence each time.

How can RxSwift operators catch all errors and emit a failure result at the end, but NOT dispose the entire sequence on first error?

Answer 1

Yes, it's a pain. I've thought about the idea of making a new library where the grammar doesn't require the stream to end on an error, but trying to reproduce the entire Rx ecosystem for it seems pointless.

There are reactive libraries that allow you to specify Never as the error type (meaning an error can't be emitted at all,) and in RxCocoa you can use Driver (which can't error) but you are still left with the whole Result dance. "Monads in my Monads!".

To deal with it properly, you need a set of Monad transformers. With these, you can do all the mapping/flatMapping you want and not worry about looking at the errors until the very end.

Answer 2

The first trick to come to mind is using materialize. This would convert every Observable<T> to Observable<Event<T>>, so an Error would just be a .next(.error(Error)) and won't cause the termination of the sequence.

in this specific case, though, another trick would be needed. Putting your entire "trigger" chain within a flatMap, as well, and materializeing that specific piece. This is needed because a materialized sequence can still complete, which would cause a termination in case of a regular chain, but would not terminate a flatMapped chain (as complete == successfully done, inside a flatMap).

inputRelay
    .flatMapLatest { val in
        return Observable.just(val)
            .map { value -> Int in
                if value == 1 { throw SomeError.randomError }
                return value + value
            }
            .flatMap { value in
                return Observable<String>.just("hey\(value)")
            }
            .materialize()
    }
    .debug("k")
    .subscribe()

    inputRelay.accept(1)
    inputRelay.accept(2)
    inputRelay.accept(3)
    inputRelay.accept(4)

This will output the following for k :

k -> subscribed
k -> Event next(error(randomError))
k -> Event next(next(hey4))
k -> Event next(completed)
k -> Event next(next(hey6))
k -> Event next(completed)
k -> Event next(next(hey8))
k -> Event next(completed)

Now all you have to do is filter just the "next" events from the materialized sequence.

If you have RxSwiftExt, you can simply use the errors() and elements() operators:

stream.elements()
    .debug("elements")
    .subscribe()

stream.errors()
    .debug("errors")
    .subscribe()

This will provide the following output:

errors -> Event next(randomError)
elements -> Event next(hey4)
elements -> Event next(hey6)
elements -> Event next(hey8)

When using this strategy, don't forget adding share() after your flatMap, so many subscriptions don't cause multiple pieces of processing.

You can read more about why you should use share in this situation here: http://adamborek.com/how-to-handle-errors-in-rxswift/

Hope this helps!