RxSwift, Share + retry mechanism

1
votes

I have a network request that can Succeed or Fail

I have encapsulated it in an observable. I have 2 rules for the request

1) There can never be more then 1 request at the same time

-> there is a share operator i can use for this

2) When the request was Succeeded i don't want to repeat the same request again and just return the latest value

-> I can use shareReplay(1) operator for this

The problem arises when the request fails, the shareReplay(1) will just replay the latest error and not restart the request again.

The request should start again at the next subscription.

Does anyone have an idea how i can turn this into a Observable chain?

// scenario 1
let obs: Observable<Int> = request().shareReplay(1)
// outputs a value
obs.subscribe()
// does not start a new request but outputs the same value as before
obs.subscribe() 

// scenario 2 - in case of an error
let obs: Observable<Int> = request().shareReplay(1)
// outputs a error
obs.subscribe() 
// does not start a new request but outputs the same value as before, but in this case i want it to start a new request
obs.subscribe()

This seems to be a exactly doing what i want, but it consists of keeping state outside the observable, anyone know how i can achieve this in a more Rx way?

enum Err: Swift.Error {
    case x
}

enum Result<T> {
    case value(val: T)
    case error(err: Swift.Error)
}

func sample() {

    var result: Result<Int>? = nil
    var i = 0

    let intSequence: Observable<Result<Int>> = Observable<Int>.create { observer in

        if let result = result {
            if case .value(let val) = result {
                return Observable<Int>.just(val).subscribe(observer)
            }
        }
        print("do work")
        delay(1) {
            if i == 0 {
                observer.onError(Err.x)
            } else {
                observer.onNext(1)
                observer.onCompleted()
            }
            i += 1
        }
        return Disposables.create {}
        }
        .map { value -> Result<Int> in Result.value(val: value) }
        .catchError { error -> Observable<Result<Int>> in
            return .just(.error(err: error))
        }
        .do(onNext: { result = $0 })
        .share()

    _ = intSequence
        .debug()
        .subscribe()

    delay(2) {
        _ = intSequence
            .debug()
            .subscribe()

        _ = intSequence
            .debug()
            .subscribe()
    }

    delay(4) {
        _ = intSequence
            .debug()
            .subscribe()
    }
}


sample()

it only generates work when we don't have anything cached, but thing again we need to use side effects to achieve the desired output

2
swiftsharerx-swift

2 Answers

5
votes

As mentioned earlier, RxSwift errors need to be treated as fatal errors. They are errors your stream usually cannot recover from, and usually errors that would not even be user facing.

For that reason - a stream that emits an .error or .completed event, will immediately dispose and you won't receive any more events there.

There are two approaches to tackling this:

  1. Using a Result type like you just did
  2. Using .materialize() (and .dematerialize() if needed). These first operator will turn your Observable<Element> into a Observable<Event<Element>>, meaning instead of an error being emitted and the sequence terminated, you will get an element that tells you it was an error event, but without any termination.

You can read more about error handling in RxSwift in Adam Borek's great blog post about this: http://adamborek.com/how-to-handle-errors-in-rxswift/

1
votes

If an Observable sequence emits an error, it can never emit another event. However, it is a fairly common practice to wrap an error-prone Observable inside of another Observable using flatMap and catch any errors before they are allowed to propagate through to the outer Observable. For example:

safeObservable
    .flatMap {
        Requestor
            .makeUnsafeObservable()
            .catchErrorJustReturn(0)
    }
    .shareReplay(1)
    .subscribe()