Reactive: Missing RetryUntil/RetryWhen

[JanEggers]

i tried to implement RetryUntil in my code with:

public static Task<T> RetryUntil<T,TOther>(this object any, IObservable<T> obs, IObservable<TOther> end)
        {
            using ( var cts = new CancellationTokenSource())
            using (end.Subscribe(v => cts.Cancel(true)))
            {
                return obs
                    .Retry()
                    .TakeUntil(end)
                    .ToTask(cts.Token);
            }
        }

but that gives me an invalidOperationException: "Sequence Contains no Elements". if end publishes a value.

Is there any way i can get a TaskCancellation Exception.

Also RetryWhen is missing from Rx.net and id like to retry after some amount of time rather than instant. How can i do that without RetryWhen?

Is there any reason the above methods are not included? / may i submit a pr?

[akarnokd]

You could emulate RetryWhen via existing operators and subjects (although the most reliable way would be a custom IObservable implementation):

public static IObservable<T> RetryWhen<T, U>(this IObservable<T> source,
    Func<IObservable<Exception>, IObservable<U>> handler)
{
    return Observable.Defer(() =>
    {
        var errorSignal = new Subject<Exception>();
        var retrySignal = handler(errorSignal);
        var sources = new BehaviorSubject<IObservable<T>>(source);

        return Observable.Using(
            () => retrySignal.Select(s => source).Subscribe(sources),
            r => sources.Select(src =>
                src.Do(v => { }, e => errorSignal.OnNext(e), () => errorSignal.OnCompleted())
                .OnErrorResumeNext(Observable.Empty<T>())
            ).Concat()
        )
        ;
    });
}

Example:

int[] count = { 3 };

Observable.Defer(() =>
{
    if (count[0]-- == 0)
    {
        return Observable.Return("Success");
    }
    return Observable.Throw<String>(new Exception());
})
.RetryWhen(
    f => f.SelectMany(e =>
    {
        Console.WriteLine("Retrying...");
        return Observable.Timer(TimeSpan.FromSeconds(1));
    })
)
.Subscribe(Console.WriteLine, Console.WriteLine, () => Console.WriteLine("Done"));

Thread.Sleep(5000);

Is there any reason the above methods are not included?

Some ReactiveX libraries evolve faster than others and not all features are available everywhere. With C# and extension methods, you are in a much better position to create your own custom operators.

[JanEggers]

thx for explaining

[quinmars]

PR #486 adds a RetryWhen operator and is already merged. Can this issue be closed?

[JanEggers]

yup thx