Write an Rx “RetryAfter” extension method

by

The key to this implementation of a back off retry is deferred observables. A deferred observable won’t execute its factory until someone subscribes to it. And it will invoke the factory for each subscription, making it ideal for our retry scenario.

Assume we have a method which triggers a network request.

public IObservable<WebResponse> SomeApiMethod() { ... }

For the purposes of this little snippet, let’s define the deferred as source

var source = Observable.Defer(() => SomeApiMethod());

Whenever someone subscribes to source it will invoke SomeApiMethod and launch a new web request. The naive way to retry it whenever it fails would be using the built in Retry operator.

source.Retry(4)

That wouldn’t be very nice to the API though and it’s not what you’re asking for. We need to delay the launching of requests in between each attempt. One way of doing that is with a delayed subscription.

Observable.Defer(() => source.DelaySubscription(TimeSpan.FromSeconds(1))).Retry(4)

That’s not ideal since it’ll add the delay even on the first request, let’s fix that.

int attempt = 0;
Observable.Defer(() => { 
   return ((++attempt == 1)  ? source : source.DelaySubscription(TimeSpan.FromSeconds(1)))
})
.Retry(4)
.Select(response => ...)

Just pausing for a second isn’t a very good retry method though so let’s change that constant to be a function which receives the retry count and returns an appropriate delay. Exponential back off is easy enough to implement.

Func<int, TimeSpan> strategy = n => TimeSpan.FromSeconds(Math.Pow(n, 2));

((++attempt == 1)  ? source : source.DelaySubscription(strategy(attempt - 1)))

We’re almost done now, we just need to add a way of specifying for which exceptions we should retry. Let’s add a function that given an exception returns whether or not it makes sense to retry, we’ll call it retryOnError.

Now we need to write some scary looking code but bear with me.

Observable.Defer(() => {
    return ((++attempt == 1)  ? source : source.DelaySubscription(strategy(attempt - 1)))
        .Select(item => new Tuple<bool, WebResponse, Exception>(true, item, null))
        .Catch<Tuple<bool, WebResponse, Exception>, Exception>(e => retryOnError(e)
            ? Observable.Throw<Tuple<bool, WebResponse, Exception>>(e)
            : Observable.Return(new Tuple<bool, WebResponse, Exception>(false, null, e)));
})
.Retry(retryCount)
.SelectMany(t => t.Item1
    ? Observable.Return(t.Item2)
    : Observable.Throw<T>(t.Item3))

All of those angle brackets are there to marshal an exception for which we shouldn’t retry past the .Retry(). We’ve made the inner observable be an IObservable<Tuple<bool, WebResponse, Exception>> where the first bool indicates if we have a response or an exception. If retryOnError indicates that we should retry for a particular exception the inner observable will throw and that will be picked up by the retry. The SelectMany just unwraps our Tuple and makes the resulting observable be IObservable<WebRequest> again.

See my gist with full source and tests for the final version. Having this operator allows us to write our retry code quite succinctly 

Observable.Defer(() => SomApiMethod())
  .RetryWithBackoffStrategy(
     retryCount: 4, 
     retryOnError: e => e is ApiRetryWebException
  )

More Related Contents:

  1. ObserveOn and SubscribeOn – where the work is being done
  2. How to throttle event stream using RX?
  3. How can I implement an exhaustMap handler in Rx.Net?
  4. Why are Subjects not recommended in .NET Reactive Extensions?
  5. How can I see what my reactive extensions query is doing?
  6. Good introduction to the .NET Reactive Framework [closed]
  7. With Rx, how do I ignore all-except-the-latest value when my Subscribe method is running
  8. Does reactive extensions support rolling buffers?
  9. Rx: How can I respond immediately, and throttle subsequent requests
  10. A way to push buffered events in even intervals
  11. What are the default Schedulers for each observable operator?
  12. How can I alternately buffer and flow a live data stream in Rx
  13. What’s a good way to run periodic tasks using Rx, with a single concurrent execution restriction?
  14. How to call back async function from Rx Subscribe?
  15. Reactive Extensions for .NET (Rx): Take action once all events are completed
  16. Pause and Resume Subscription on cold IObservable
  17. How to check if element exist or not without using try-catch
  18. How can I update the current line in a C# Windows Console App?
  19. How to serialize a Dictionary as part of its parent object using Json.Net
  20. Lifetime of ASP.NET Static Variable
  21. FTPS (FTP over SSL) in C#
  22. Drawing glitches when using CreateGraphics rather than Paint event handler for custom drawing
  23. Fast creation of objects instead of Activator.CreateInstance(type)
  24. Sending Windows key using SendKeys
  25. Why can’t we assign a foreach iteration variable, whereas we can completely modify it with an accessor?
  26. Using a .NET DLL in Node.js / serverside javascript
  27. Using a Scoped service in a Singleton in an Asp.Net Core app
  28. How to update querystring in C#?
  29. Why is (double)0.6f > (double)(6/10f)?
  30. How to find path of active app.config file?

Leave a Comment