Task sequencing and re-entracy
Microsoft's Rx does provide an easy way to do this kind of thing. Here's a simple (perhaps overly simple) way of doing it:
var subject = new BehaviorSubject<int>(0);
IDisposable subscription =
subject
.Scan((x0, x1) =>
{
Console.WriteLine($"previous value {x0}");
return x1;
})
.Skip(1)
.Subscribe(x => Console.WriteLine($"current value {x}\r\n"));
subject.OnNext(1000);
subject.OnNext(900);
subject.OnNext(800);
Console.WriteLine("\r\nPress any key to continue to test #2...\r\n");
Console.ReadLine();
subject.OnNext(200);
subject.OnNext(100);
Console.WriteLine("\r\nPress any key to exit...");
Console.ReadLine();
The output I get is this:
previous value 0 current value 1000 previous value 1000 current value 900 previous value 900 current value 800 Press any key to continue to test #2... previous value 800 current value 200 previous value 200 current value 100 Press any key to exit...
It's easy to cancel at any time by calling subscription.Dispose().
Error handling in Rx is generally a little more bespoke than normal. It's not just a matter of throwing a try/catch around things. You also can repeat steps that error with a Retry operator in the case of things like IO errors.
In this circumstance, because I've used a BehaviorSubject (which repeats its last value whenever it is subscribed to) you can easily just resubscribe using a Catch operator.
var subject = new BehaviorSubject<int>(0);
var random = new Random();
IDisposable subscription =
subject
.Select(x =>
{
if (random.Next(10) == 0)
throw new Exception();
return x;
})
.Catch<int, Exception>(ex => subject.Select(x => -x))
.Scan((x0, x1) =>
{
Console.WriteLine($"previous value {x0}");
return x1;
})
.Skip(1)
.Subscribe(x => Console.WriteLine($"current value {x}\r\n"));
Now with the .Catch<int, Exception>(ex => subject.Select(x => -x)) it inverts the value of the query should an exception be raised.
A typical output may be like this:
previous value 0 current value 1000 previous value 1000 current value 900 previous value 900 current value 800 Press any key to continue to test #2... previous value 800 current value -200 previous value -200 current value -100 Press any key to exit...
Note the -ve numbers in the second half. An exception was handled and the query was able to continue.
Here is a solution that is worse on every aspect compared to the accepted answer, except from being thread-safe (which is not a requirement of the question). Disadvantages:
- All lambdas are executed asynchronously (there is no fast path).
- The
executeOnCurrentContextconfiguration effects all lambdas (it's not a per-lambda configuration).
This solution uses as processing engine an ActionBlock from the TPL Dataflow library.
public class AsyncOp<T>
{
private readonly ActionBlock<Task<Task<T>>> _actionBlock;
public AsyncOp(bool executeOnCurrentContext = false)
{
var options = new ExecutionDataflowBlockOptions();
if (executeOnCurrentContext)
options.TaskScheduler = TaskScheduler.FromCurrentSynchronizationContext();
_actionBlock = new ActionBlock<Task<Task<T>>>(async taskTask =>
{
try
{
taskTask.RunSynchronously();
await await taskTask;
}
catch { } // Ignore exceptions
}, options);
}
public Task<T> RunAsync(Func<Task<T>> taskFactory)
{
var taskTask = new Task<Task<T>>(taskFactory);
if (!_actionBlock.Post(taskTask))
throw new InvalidOperationException("Not accepted"); // Should never happen
return taskTask.Unwrap();
}
}
I almost forgot it's possible to construct a Task manually, without starting or scheduling it. Then, "Task.Factory.StartNew" vs "new Task(...).Start" put me back on track. I think this is one of those few cases when the Task<TResult> constructor may actually be useful, along with nested tasks (Task<Task<T>>) and Task.Unwrap():
// AsyncOp
class AsyncOp<T>
{
Task<T> _pending = Task.FromResult(default(T));
public Task<T> CurrentTask { get { return _pending; } }
public Task<T> RunAsync(Func<Task<T>> handler, bool useSynchronizationContext = false)
{
var pending = _pending;
Func<Task<T>> wrapper = async () =>
{
// await the prev task
var prevResult = await pending;
Console.WriteLine("\nprev task result: " + prevResult);
// start and await the handler
return await handler();
};
var task = new Task<Task<T>>(wrapper);
var inner = task.Unwrap();
_pending = inner;
task.RunSynchronously(useSynchronizationContext ?
TaskScheduler.FromCurrentSynchronizationContext() :
TaskScheduler.Current);
return inner;
}
}
The output:
Test #1... prev task result: 0 this task arg: 1000 prev task result: 1000 this task arg: 900 prev task result: 900 this task arg: 800 Press any key to continue to test #2... prev task result: 800 this task arg: 100 prev task result: 100 this task arg: 200
It's now also very easy to make AsyncOp thread-safe by adding a lock to protect _pending, if needed.
Updated, this has been further improved with cancel/restart logic.