Rxswift: Is there a timeout operator?

[frogcjn]

Is there a timeout operator?
http://reactivex.io/documentation/operators/timeout.html

AVAsset has a function called loadValuesAsynchronouslyForKeys. I'm going to use RxSwift to deal with it. But if an HTTP resource cannot be got, this method will wait until 60 seconds to return. So there is a need to limit the time for this async fun to return.

[frogcjn]

I can use

        .takeUntil(timer(2.0, MainScheduler.sharedInstance))

but there is no error will return. Just complete.

[IanKeen]

You could create a 'timeout' observable that simply errors after a set duration and merge that with your request.. possibly not ideal but you could abstract it into something nice

[frogcjn]

@IanKeen , this request is system triggered, which I cannot change the timeout property of the NSURLRequest.

[IanKeen]

right, but you can combine it with another observable which you do control the timeout of (unsure if this is even the best way... just a suggestion on how you might tackle the problem)

[frogcjn]

@IanKeen

like

    .takeUntil(timer(2.0, MainScheduler.sharedInstance))

but there is no error will return. Just complete.

So I should have a single result ensure operator like this

    var subject: Observable<Element> = ...
    subject
            .takeUntil(timer(2.0, MainScheduler.sharedInstance))
            .toArray()
            .map { array -> Element in
                if array.count > 1{ throw CommonError.MoreThanOneResult}
                if array.count < 1{ throw CommonError.NoResult}
                return array.first!
            }

[IanKeen]

I was thinking perhaps something like

        return create { observer in
            let delayTime = dispatch_time(DISPATCH_TIME_NOW, Int64(2.0 * Double(NSEC_PER_SEC)))
            dispatch_after(delayTime, dispatch_get_main_queue()) {
                observer.onError(error)
            }
        }

[frogcjn]

@IanKeen

Yes, you're right.

There also needs a flag to mark whether the function was returned by system.

    return create { observer in
        var returned = false
        let delayTime = dispatch_time(DISPATCH_TIME_NOW, Int64(2.0 * Double(NSEC_PER_SEC)))
        dispatch_after(delayTime, dispatch_get_main_queue()) {
            if !returned { observer.onError(error) }
        }
        someSystemAsyncFunc{
              returned = true
               observer.onNext(...)
               observer.onCompleted()
        }
    }

[kzaher]

Hi,

I believe timeout should be something like

sequenceOf(originalObservable, failWith(testError).delaySubscription(timeoutInterval, MainScheduler.sharedInstance)).merge()

:)

We don't currently have timeout operator, but that is a good idea of a useful operator we would want to have :)

[frogcjn]

@kzaher Thanks! Your solution is perfect.

[frogcjn]

@kzaher
I have tested, this solution will lead to a result that there is always an timeout error.

merge will not be completed after first one completed. "merge" will wait the second error to complete.

[kzaher]

Ah, my bad.

What about this

originalObservable.amb(failWith(testError).delaySubscription(timeoutInterval, MainScheduler.sharedInstance))

:)

[frogcjn]

@kzaher
Yes, that works. Thanks again!

[fpillet]

Think you can close this one now

[bontoJR]

:+1: