Create observable from periodic async request
Here is my take on this problem:
Update: I was able to simplify greatly my suggested solution by borrowing ideas from Enigmativity's answer. The Observable.StartAsync
method handles the messy business of cancellation automatically, and the requirement of non-overlapping execution can be enforced simply by using a SemaphoreSlim
.
/// <summary>
/// Creates an observable sequence containing the results of an asynchronous
/// function that is invoked periodically and manually. Overlapping invocations
/// are prevented. Timer ticks that would cause overlapping are ignored.
/// Manual invocations cancel previous invocations, and restart the timer.
/// </summary>
public static IObservable<T> PeriodicAndManual<T>(
Func<bool, CancellationToken, Task<T>> functionAsync,
TimeSpan period,
out Action manualInvocation)
{
// Arguments validation omitted
var manualSubject = new Subject<bool>();
manualInvocation = () => manualSubject.OnNext(true);
var semaphore = new SemaphoreSlim(1);
return Observable
.Interval(period)
.Select(_ => false) // Not manual
.Merge(manualSubject)
.TakeUntil(isManual => isManual) // Stop on first manual
.Repeat() // ... and restart the timer
.Prepend(false) // Skip the initial interval delay
.Scan(seed: (
// Both representations of an operation are needed
// The Observable provides automatic cancellation on unsubscription
// The Task maintains the IsCompleted state
Operation: (IObservable<T>)null,
AsTask: Task.FromResult(default(T))
), accumulator: (previous, isManual) =>
{
// Start a new operation only if the previous operation is completed,
// or if the call is manual. Otherwise return the previous operation.
if (!previous.AsTask.IsCompleted && !isManual) return previous;
// Start a new operation as hot observable
var operation = Observable.StartAsync(async ct =>
{
await semaphore.WaitAsync(ct); // Ensure no overlapping
try { return await functionAsync(isManual, ct); }
finally { semaphore.Release(); }
}, Scheduler.Immediate); // Propagate the task status synchronously
return (operation, operation.ToTask());
})
.Select(entry => entry.Operation) // Discard the AsTask representation
.DistinctUntilChanged() // Ignore duplicate operations
.Switch(); // Cancel pending operations and ignore them
}
The out Action manualInvocation
argument is the mechanism that triggers a manual invocation.
Usage example:
int ticks = 0;
var subscription = PeriodicAndManual(async (isManual, token) =>
{
var id = $"{++ticks} " + (isManual ? "manual" : "periodic");
Console.WriteLine($"{DateTime.Now:HH:mm:ss.fff} Begin {id}");
await Task.Delay(500, token);
return id;
}, TimeSpan.FromMilliseconds(1000), out var manualInvocation)
.Do(x => Console.WriteLine($"{DateTime.Now:HH:mm:ss.fff} Received #{x}"))
.Subscribe();
await Task.Delay(3200);
manualInvocation();
await Task.Delay(200);
manualInvocation();
await Task.Delay(3200);
subscription.Dispose();
Output:
19:52:43.684 Begin 1 periodic
19:52:44.208 Received #1 periodic
19:52:44.731 Begin 2 periodic
19:52:45.235 Received #2 periodic
19:52:45.729 Begin 3 periodic
19:52:46.232 Received #3 periodic
19:52:46.720 Begin 4 periodic
19:52:46.993 Begin 5 manual
19:52:47.220 Begin 6 manual
19:52:47.723 Received #6 manual
19:52:48.223 Begin 7 periodic
19:52:48.728 Received #7 periodic
19:52:49.227 Begin 8 periodic
19:52:49.730 Received #8 periodic
19:52:50.226 Begin 9 periodic
The technique of using the Scan
and the DistinctUntilChanged
operators in order to drop elements while the previous asynchronous operation is running, is borrowed from this question.