I wrote a simple utility for Retry and CircuitBreaker patterns. I implemented it by using ReactiveX because I thought it might actually be the perfect framework for it. I know that there is a project called Polly but it didn't convince me and I wanted to try Rx out.
My little tool can basically do two things:
- It can retry an action as many times as I want and wait in between the specified amount of time.
- It can break retries of one or multiple retry handlers with a circuit-breaker.
Retry
The Retry class implements the IObservable<Exception> interface. It notifies its observers (if any) every time the action that might fail actually failed by sending it the current exception via OnNext. Retry never calls OnCompleted because it actually never completes and we don't want to unsubscribe any observables in case we reuse the same Retry another time.
Internally it maintains a HashSet of CircutBreakers that provide a Token property that is necessary to cancel the Retry.
I'm not sure whether I should actually collect the exceptions an throw the AggregateException at the end or just leave the method with a failed Task and let other observer(s) take care of (maybe log) the exception(s).
public class Retry : IObservable<Exception>
{
private readonly IEnumerable<TimeSpan> _delays;
private readonly Subject<Exception> _subject;
private readonly HashSet<CircutBraker> _circutBreakers;
private Retry(IEnumerable<TimeSpan> delays)
{
// Deliberately not materializing the delays. We don't know whether this is not infinite.
_delays = delays;
_subject = new Subject<Exception>();
_circutBreakers = new HashSet<CircutBraker>();
}
public IDisposable Subscribe(IObserver<Exception> exceptionObserver)
{
if (exceptionObserver is CircutBraker circutBreaker)
{
_circutBreakers.Add(circutBreaker);
var circutBreakerUnsubscriber = _subject.Subscribe(exceptionObserver);
return Disposable.Create(() =>
{
_circutBreakers.Remove(circutBreaker);
circutBreakerUnsubscriber.Dispose();
});
}
else
{
return _subject.Subscribe(exceptionObserver);
}
}
public static Retry Create(IEnumerable<TimeSpan> delays)
{
return new Retry(delays);
}
public async Task<T> TryExecuteAsync<T>(Func<CancellationToken, Task<T>> action, CancellationToken cancellationToken)
{
var exceptions = new List<Exception>();
foreach (var delay in _delays)
{
ThrowIfCancellationRequested();
try
{
return await action(cancellationToken);
}
catch (Exception ex)
{
exceptions.Add(ex);
_subject.OnNext(ex);
// Don't delay if cancelled but return immediately.
ThrowIfCancellationRequested();
await Task.Delay(delay);
}
}
throw new AggregateException(exceptions);
void ThrowIfCancellationRequested()
{
foreach (var cb in _circutBreakers)
{
cb.Token.ThrowIfCancellationRequested();
}
}
}
}
CircutBreaker v1
The CircutBreaker class is implemented on top of the IObserver<Exception> interface and the main logic is created inside the constructor. Currently it supports just two states: Closed and Opened - because I don't have any scenario yet where I could use the third state and check if the error is gone (I might implement it later) - thus the HalfOpen case is empty.
When the number of failures exceeds the allowed count within the specified interval the CancellationTokenSource is cancelled. Retry watches for this token and reacts accordingly by checking the Token property - this allows me to cancel all retries immediately.
public class CircutBreaker : IObserver<Exception>, IDisposable
{
private readonly IObserver<Exception> _observer;
private DateTime _lastFailureOn;
private int _failureCount;
private CircutBrakerState _state = CircutBrakerState.Closed;
private CancellationTokenSource _cancellationTokenSource = new CancellationTokenSource();
private CircutBreaker(int maxFailureCount, TimeSpan maxFailureInterval, Func<Exception, bool> filter = null)
{
_observer = Observer.Create<Exception>(exception =>
{
if (filter?.Invoke(exception) == false)
{
return;
}
var now = DateTime.UtcNow;
switch (_state)
{
case CircutBrakerState.Closed:
{
var isFailureWithinInterval = now - _lastFailureOn < maxFailureInterval;
var canCountFailure = isFailureWithinInterval || _failureCount == 0;
_failureCount = canCountFailure ? _failureCount + 1 : 0;
if (_failureCount == maxFailureCount)
{
_state = CircutBrakerState.Open;
_cancellationTokenSource.Cancel();
Debug.WriteLine("Circut broken!");
}
_lastFailureOn = now;
}
break;
case CircutBrakerState.HalfOpen:
// I currently don't have any use case for this.
break;
case CircutBrakerState.Open:
// There is nothing to do when Open.
break;
}
});
}
public CircutBrakerState State => _state;
public CancellationToken Token => _cancellationTokenSource.Token;
public void OnNext(Exception exception) => _observer.OnNext(exception);
public void OnCompleted() => _observer.OnCompleted();
public void OnError(Exception exception) => _observer.OnError(exception);
public void Reset()
{
_lastFailureOn = DateTime.MinValue;
_failureCount = 0;
_state = CircutBrakerState.Closed;
_cancellationTokenSource.Dispose();
_cancellationTokenSource = new CancellationTokenSource();
}
public void Dispose() => _cancellationTokenSource.Dispose();
public static CircutBreaker Create(int maxFailureCount, TimeSpan maxFailureInterval, Func<Exception, bool> filter = null)
{
return new CircutBreaker(maxFailureCount, maxFailureInterval, filter);
}
}
CircutBreaker v2
There is also an alternative CircutBreaker constructor which uses a new extension to open the circut:
private CircutBreaker(int maxFailureCount, TimeSpan maxFailureInterval, Func<Exception, bool> filter = null)
{
var openFuse = Observer.Create<Exception>(exception =>
{
switch (_state)
{
case CircutBrakerState.Closed:
_state = CircutBrakerState.Open;
_cancellationTokenSource.Cancel();
Debug.WriteLine("Circut broken!");
break;
case CircutBrakerState.HalfOpen:
// I currently don't have any use case for this.
break;
case CircutBrakerState.Open:
// There is nothing to do when Open.
break;
}
});
_observer = new Subject<Exception>();
((IObservable<Exception>)_observer)
.Where(ex => filter == null || filter(ex))
.Fuse(maxFailureCount, maxFailureInterval)
.Subscribe(openFuse);
}
I named the extension Fuse because it lets only the last item through that occurred within the specified interval:
public static class ObservableExtensions
{
public static IObservable<T> Fuse<T>(this IObservable<T> observable, int count, TimeSpan interval)
{
var currentCount = 0;
var lastCountOn = DateTime.MinValue;
var fuse = new Subject<T>();
observable.Subscribe(Observer.Create<T>(item =>
{
if (currentCount == 0 || DateTime.UtcNow - lastCountOn < interval)
{
currentCount++;
if (currentCount == count)
{
fuse.OnNext(item);
currentCount = 0;
}
}
else
{
currentCount = 0;
}
lastCountOn = DateTime.UtcNow;
}));
return fuse;
}
}
I wrote this one in order to remove the fuse logic out of the CircuitBreaker. Do you think it is a good choice? This is why I added the tag comparative-review.
The states are defined in an enum:
public enum CircutBreakerState
{
Closed,
HalfOpen,
Open,
}
I also crated an observer-logger that I can use to log exceptions or display the message to the console:
public class ExceptionLogger
{
public static IObserver<Exception> Create(Action<Exception> logException)
{
return Observer.Create(logException);
}
}
Example
I tested it by creating two Retry objects and one CircuitBreaker. I then attached it to both retries and executed a method that fails. Additionally, I attached a logger to each retry.
var retry1 = Retry.Create(new[]
{
TimeSpan.FromSeconds(3),
TimeSpan.FromSeconds(3),
TimeSpan.FromSeconds(3),
TimeSpan.FromSeconds(3)
});
var retry2 = Retry.Create(new[]
{
TimeSpan.FromSeconds(3),
TimeSpan.FromSeconds(3),
TimeSpan.FromSeconds(3),
TimeSpan.FromSeconds(3)
});
var circutBreaker = CircutBreaker.Create(2, TimeSpan.FromSeconds(6));
retry1.Subscribe(ExceptionLogger.Create(ex => Console.WriteLine("Log: " + ex.Message)));
retry2.Subscribe(ExceptionLogger.Create(ex => Console.WriteLine("Log: " + ex.Message)));
retry1.Subscribe(circutBreaker);
retry2.Subscribe(circutBreaker);
var task1 = retry1.TryExecuteAsync(async (cancellationToken) =>
{
Console.WriteLine("1st method called.");
await Task.Delay(TimeSpan.FromSeconds(0.5));
throw new Exception("1st method failed.");
return "Test";
}, CancellationToken.None);
var task2 = retry2.TryExecuteAsync(async (cancellationToken) =>
{
Console.WriteLine("2nd method called.");
await Task.Delay(TimeSpan.FromSeconds(0.5));
throw new Exception("2nd method failed.");
return "Test";
}, CancellationToken.None);
Task.WaitAll(new[] { task1, task2 });
The output is:
1st method called.
2nd method called.
Log: 2nd method failed.
Log: 1st method failed.
Circut broken!
It works very nice but I think it can still be made better and more reactive.
