diff --git a/TaskFlow.Extensions.Microsoft.Logging.Tests/LoggingTaskSchedulerExtensionsFixture.cs b/TaskFlow.Extensions.Microsoft.Logging.Tests/LoggingTaskSchedulerExtensionsFixture.cs
index 92509fc..db8805d 100644
--- a/TaskFlow.Extensions.Microsoft.Logging.Tests/LoggingTaskSchedulerExtensionsFixture.cs
+++ b/TaskFlow.Extensions.Microsoft.Logging.Tests/LoggingTaskSchedulerExtensionsFixture.cs
@@ -29,7 +29,7 @@ public async Task WithLogging_LogsFullLifecycleAtTraceByDefault()
{
_taskFlow = new TaskFlow();
var logger = new RecordingLogger(LogLevel.Trace);
- Assert.That(await _taskFlow.WithLogging(logger).WithOperationName("answer").Enqueue(() => 42), Is.EqualTo(42));
+ Assert.That(await _taskFlow.WithOperationName("answer").WithLogging(logger).Enqueue(() => 42), Is.EqualTo(42));
Assert.That(logger.Entries.Select(x => x.EventId.Id), Is.EqualTo(new[] { EnqueuedEventId, StartedEventId, SucceededEventId, FinishedEventId }));
Assert.That(logger.Entries, Has.All.Property(nameof(LogEntry.Level)).EqualTo(LogLevel.Trace));
Assert.That(logger.Entries.Select(x => x.Message), Has.All.Contains("operation 1"));
diff --git a/TaskFlow.Extensions.Microsoft.Logging/LoggingTaskSchedulerExtensions.cs b/TaskFlow.Extensions.Microsoft.Logging/LoggingTaskSchedulerExtensions.cs
index 15a129e..fa2140f 100644
--- a/TaskFlow.Extensions.Microsoft.Logging/LoggingTaskSchedulerExtensions.cs
+++ b/TaskFlow.Extensions.Microsoft.Logging/LoggingTaskSchedulerExtensions.cs
@@ -14,80 +14,76 @@ public static class LoggingTaskSchedulerExtensions
private static readonly EventId FailedEvent = new EventId(0x5446_0005, "TaskFlowOperationFailed");
private static readonly EventId FinishedEvent = new EventId(0x5446_0006, "TaskFlowOperationFinished");
- /// Wraps a scheduler with configurable structured lifecycle logging.
+ /// Registers structured enqueue and execution lifecycle logging for every scheduled operation.
/// The scheduler whose operations will be logged.
/// The logger that receives lifecycle events.
///
/// An optional callback that configures event levels. When omitted, every lifecycle event uses
/// .
///
- /// An that logs the lifecycle of every enqueued operation.
+ /// A new immutable scheduler snapshot that logs operation lifecycles.
///
- /// Thrown when or is null.
+ /// or is null.
///
///
///
- /// The decorator emits enqueue, start, cancellation-request, success or failure, and finish events. Events
- /// contain structured operation ID, optional operation name, result type, and elapsed-duration fields where
- /// applicable. Failure events include the operation exception.
+ /// The compound middleware emits enqueue, start, cancellation-request, success or failure, and finish events.
+ /// Every event carries an increasing operation ID and the visible when
+ /// this logging registration is created. Result type and elapsed duration are included where applicable.
///
///
- /// is checked before logging each event and before optional timing
- /// work is started. Disabled events do not call .
+ /// Call before
+ /// WithLogging. Metadata is forward-scoped, so an operation name registered later does not retroactively
+ /// change this logging registration.
///
///
- /// Cancellation requests are observed synchronously from the cancellation token. This event does not mean
- /// that cancellation was accepted or that the operation ultimately completed as canceled.
- ///
- ///
- /// Place WithOperationName outside this decorator, for example
- /// scheduler.WithLogging(logger).WithOperationName("Import"), so logging can observe the annotation.
+ /// is checked before each event and before optional timing work begins.
+ /// Disabled events do not invoke .
+ /// Cancellation-request logging observes the submitting caller's token and does not imply that the operation
+ /// ultimately completes as canceled. Logging never suppresses or replaces the operation outcome.
///
+ /// The returned pipeline snapshot is non-owning and does not dispose the underlying scheduler or logger.
///
///
///
/// var scheduler = taskFlow
- /// .WithLogging(logger, options =>
+ /// .WithOperationName("imports.run")
+ /// .WithLogging(logger, options =>
/// {
+ /// options.StartedLogLevel = LogLevel.Information;
/// options.FailedLogLevel = LogLevel.Error;
/// options.FinishedLogLevel = LogLevel.Debug;
- /// })
- /// .WithOperationName("Import");
+ /// });
///
- /// await scheduler.Enqueue(() => ImportAsync());
+ /// await scheduler.Enqueue(token => ImportAsync(token));
///
///
public static ITaskScheduler WithLogging(this ITaskScheduler taskScheduler, ILogger logger, Action? configure = null)
{
Argument.NotNull(taskScheduler);
Argument.NotNull(logger);
-
var options = new TaskFlowLoggingOptions();
configure?.Invoke(options);
- return new LoggingTaskSchedulerWrapper(taskScheduler, logger, options);
+ return taskScheduler.UseMiddleware(new LoggingMiddleware(logger, options));
}
- private sealed class LoggingTaskSchedulerWrapper : ITaskScheduler
+ private sealed class LoggingMiddleware : ITaskSchedulerEnqueueMiddleware, ITaskSchedulerExecutionMiddleware
{
private readonly ILogger _logger;
private readonly TaskFlowLoggingOptions _options;
- private readonly ITaskScheduler _interceptedScheduler;
private long _lastOperationId;
- public LoggingTaskSchedulerWrapper(ITaskScheduler taskScheduler, ILogger logger, TaskFlowLoggingOptions options)
+ public LoggingMiddleware(ILogger logger, TaskFlowLoggingOptions options)
{
_logger = logger;
_options = options;
- _interceptedScheduler = taskScheduler.Intercept(new LoggingInterceptor(logger, options));
}
- public async Task Enqueue(Func> taskFunc, object? state, CancellationToken cancellationToken)
+ public async Task InvokeAsync(TaskSchedulerEnqueueContext context, TaskSchedulerEnqueueDelegate continuation)
{
- var operation = new LoggingOperationState(
+ var operation = context.GetOrCreateLocalState(() => new LoggingOperationState(
Interlocked.Increment(ref _lastOperationId),
- TaskSchedulerInterceptionContext.GetAnnotation(state)?.OperationName,
- taskFunc,
- state);
+ context.GetAnnotation()?.OperationName));
if (_logger.IsEnabled(_options.EnqueuedLogLevel))
{
@@ -95,7 +91,7 @@ public async Task Enqueue(Func> t
"TaskFlow operation {OperationId} ({OperationName}) enqueued", operation.OperationId, operation.OperationName);
}
- using var registration = cancellationToken.Register(() =>
+ using var registration = context.CallerCancellationToken.Register(() =>
{
if (_logger.IsEnabled(_options.CancellationRequestedLogLevel))
{
@@ -104,118 +100,81 @@ public async Task Enqueue(Func> t
}
});
- return await _interceptedScheduler.Enqueue(Execute, operation, cancellationToken).ConfigureAwait(false);
-
- static ValueTask Execute(object? operationState, CancellationToken token)
- {
- var loggingState = (LoggingOperationState)operationState!;
- return loggingState.TaskFunc(loggingState.State, token);
- }
+ return await continuation(context).ConfigureAwait(false);
}
- }
-
- private struct LoggingInterceptor : ITaskSchedulerInterceptor
- {
- private readonly ILogger _logger;
- private readonly TaskFlowLoggingOptions _options;
- private long _startTimestamp;
- public LoggingInterceptor(ILogger logger, TaskFlowLoggingOptions options)
+ public async ValueTask InvokeAsync(TaskSchedulerOperationContext context, TaskSchedulerExecutionDelegate continuation)
{
- _logger = logger;
- _options = options;
- _startTimestamp = 0;
- }
+ var operation = context.GetOrCreateLocalState(() => new LoggingOperationState(
+ Interlocked.Increment(ref _lastOperationId),
+ context.GetAnnotation()?.OperationName));
- public void OnBefore(TaskSchedulerInterceptionContext context)
- {
if (_logger.IsEnabled(_options.SucceededLogLevel) || _logger.IsEnabled(_options.FailedLogLevel) || _logger.IsEnabled(_options.FinishedLogLevel))
{
- _startTimestamp = Stopwatch.GetTimestamp();
+ operation.StartTimestamp = Stopwatch.GetTimestamp();
}
if (_logger.IsEnabled(_options.StartedLogLevel))
{
- var operation = GetLoggingState(context);
_logger.Log(_options.StartedLogLevel, StartedEvent,
"TaskFlow operation {OperationId} ({OperationName}) started", operation.OperationId, operation.OperationName);
}
- }
-
- public void OnSuccess(TaskSchedulerInterceptionContext context, TResult result)
- {
- if (_logger.IsEnabled(_options.SucceededLogLevel))
+ try
{
- var operation = GetLoggingState(context);
- _logger.Log(_options.SucceededLogLevel, SucceededEvent,
- "TaskFlow operation {OperationId} ({OperationName}) succeeded with result type {ResultType} in {ElapsedMilliseconds} ms",
- operation.OperationId, operation.OperationName, typeof(TResult).FullName, GetElapsedMilliseconds());
- }
+ TResult result;
+ try
+ {
+ result = await continuation(context).ConfigureAwait(true);
+ }
+ catch (Exception exception)
+ {
+ if (_logger.IsEnabled(_options.FailedLogLevel))
+ {
+ _logger.Log(_options.FailedLogLevel, FailedEvent, exception,
+ "TaskFlow operation {OperationId} ({OperationName}) failed in {ElapsedMilliseconds} ms",
+ operation.OperationId, operation.OperationName, operation.GetElapsedMilliseconds());
+ }
+
+ throw;
+ }
- }
+ if (_logger.IsEnabled(_options.SucceededLogLevel))
+ {
+ _logger.Log(_options.SucceededLogLevel, SucceededEvent,
+ "TaskFlow operation {OperationId} ({OperationName}) succeeded with result type {ResultType} in {ElapsedMilliseconds} ms",
+ operation.OperationId, operation.OperationName, typeof(TResult).FullName, operation.GetElapsedMilliseconds());
+ }
- public void OnError(TaskSchedulerInterceptionContext context, Exception exception)
- {
- if (_logger.IsEnabled(_options.FailedLogLevel))
- {
- var operation = GetLoggingState(context);
- _logger.Log(_options.FailedLogLevel, FailedEvent, exception,
- "TaskFlow operation {OperationId} ({OperationName}) failed in {ElapsedMilliseconds} ms",
- operation.OperationId, operation.OperationName, GetElapsedMilliseconds());
+ return result;
}
-
- }
-
- public void OnFinally(TaskSchedulerInterceptionContext context)
- {
- LogFinished(context);
- }
-
- private void LogFinished(TaskSchedulerInterceptionContext context)
- {
- if (_logger.IsEnabled(_options.FinishedLogLevel))
+ finally
{
- var operation = GetLoggingState(context);
- _logger.Log(_options.FinishedLogLevel, FinishedEvent,
- "TaskFlow operation {OperationId} ({OperationName}) finished in {ElapsedMilliseconds} ms",
- operation.OperationId, operation.OperationName, GetElapsedMilliseconds());
+ if (_logger.IsEnabled(_options.FinishedLogLevel))
+ {
+ _logger.Log(_options.FinishedLogLevel, FinishedEvent,
+ "TaskFlow operation {OperationId} ({OperationName}) finished in {ElapsedMilliseconds} ms",
+ operation.OperationId, operation.OperationName, operation.GetElapsedMilliseconds());
+ }
}
}
-
- private double GetElapsedMilliseconds()
- {
- return _startTimestamp != 0
- ? (Stopwatch.GetTimestamp() - _startTimestamp) * 1000d / Stopwatch.Frequency
- : 0d;
- }
-
- private static ILoggingOperationState GetLoggingState(TaskSchedulerInterceptionContext context)
- {
- return (ILoggingOperationState)context.State!;
- }
- }
-
- private interface ILoggingOperationState
- {
- long OperationId { get; }
- string? OperationName { get; }
}
- private sealed class LoggingOperationState : ILoggingOperationState
+ private sealed class LoggingOperationState
{
- public LoggingOperationState(long operationId, string? operationName, Func> taskFunc, object? state)
+ public LoggingOperationState(long operationId, string? operationName)
{
OperationId = operationId;
OperationName = operationName;
- TaskFunc = taskFunc;
- State = state;
}
public long OperationId { get; }
public string? OperationName { get; }
- public Func> TaskFunc { get; }
- public object? State { get; }
+ public long StartTimestamp { get; set; }
+
+ public double GetElapsedMilliseconds() => StartTimestamp == 0
+ ? 0d
+ : (Stopwatch.GetTimestamp() - StartTimestamp) * 1000d / Stopwatch.Frequency;
}
}
}
diff --git a/TaskFlow.Tests/Extensions/ExceptionTaskSchedulerExtensionsFixture.cs b/TaskFlow.Tests/Extensions/ExceptionTaskSchedulerExtensionsFixture.cs
index 02d8422..211e820 100644
--- a/TaskFlow.Tests/Extensions/ExceptionTaskSchedulerExtensionsFixture.cs
+++ b/TaskFlow.Tests/Extensions/ExceptionTaskSchedulerExtensionsFixture.cs
@@ -105,5 +105,59 @@ public void Enqueue_MultipleHandlers_ShouldExecuteAllMatchingHandlers(ITaskFlow
Assert.That(exceptions, Is.EqualTo(Generic));
exceptions.Clear();
}
+
+ [Test]
+ public void OnError_DedicatedThread_AsynchronousFailureCallbackRetainsFlowContext()
+ {
+ var flow = new DedicatedThreadTaskFlow();
+ _taskFlow = flow;
+ var operationThread = 0;
+ var callbackThread = 0;
+ SynchronizationContext? operationContext = null;
+ SynchronizationContext? callbackContext = null;
+
+ var task = flow.OnError(_ =>
+ {
+ callbackThread = Environment.CurrentManagedThreadId;
+ callbackContext = SynchronizationContext.Current;
+ })
+ .Enqueue(FailAsync);
+
+ Assert.That(async () => await task, Throws.InvalidOperationException);
+ Assert.That(callbackThread, Is.EqualTo(operationThread));
+ Assert.That(callbackContext, Is.SameAs(operationContext).And.Not.Null);
+
+ async ValueTask FailAsync(CancellationToken _)
+ {
+ await Task.Yield();
+ operationThread = Environment.CurrentManagedThreadId;
+ operationContext = SynchronizationContext.Current;
+ throw new InvalidOperationException("expected");
+ }
+ }
+
+ [TestCaseSource(typeof(TaskFlows), nameof(TaskFlows.CreateTaskFlows))]
+ public void OnError_CallbackFailureBecomesOutcomeForLaterHandlers(ITaskFlow taskFlow)
+ {
+ _taskFlow = taskFlow;
+ Exception? observed = null;
+ var scheduler = taskFlow
+ .OnError(_ => throw new NotSupportedException("replacement"))
+ .Intercept(new NoOpInterceptor())
+ .OnError(exception => observed = exception);
+
+ var task = scheduler.Enqueue(() => throw new InvalidOperationException("original"));
+
+ Assert.That(async () => await task, Throws.TypeOf().With.Message.EqualTo("replacement"));
+ Assert.That(observed, Is.TypeOf().With.Message.EqualTo("replacement"));
+ }
+
+ private readonly struct NoOpInterceptor : ITaskSchedulerInterceptor
+ {
+ public void OnBefore(TaskSchedulerInterceptionContext context) { }
+ public void OnSuccess(TaskSchedulerInterceptionContext context, TResult result) { }
+ public void OnError(TaskSchedulerInterceptionContext context, Exception exception) { }
+ public void OnFinally(TaskSchedulerInterceptionContext context) { }
+ }
}
}
diff --git a/TaskFlow.Tests/Extensions/FuturePolicyMiddlewareSeamsFixture.cs b/TaskFlow.Tests/Extensions/FuturePolicyMiddlewareSeamsFixture.cs
new file mode 100644
index 0000000..8d7209a
--- /dev/null
+++ b/TaskFlow.Tests/Extensions/FuturePolicyMiddlewareSeamsFixture.cs
@@ -0,0 +1,524 @@
+namespace TaskFlow.Tests.Extensions
+{
+ using NUnit.Framework;
+ using System.Collections.Concurrent;
+ using System.Diagnostics.CodeAnalysis;
+ using System.Threading.Tasks.Flow;
+
+ [TestFixture]
+ internal sealed class FuturePolicyMiddlewareSeamsFixture
+ {
+ [SuppressMessage("Maintainability", "CA1515:Consider making public types internal", Justification = "NUnit requires the parameter type of this public parameterized test to be publicly accessible.")]
+ public enum OwnershipCancellation
+ {
+ LocalScope,
+ SharedLane,
+ }
+
+ [Test]
+ public async Task Retry_UsesOneQueueTurnAndInvokesDownstreamOncePerAttempt()
+ {
+ var terminal = new CountingInlineTerminal();
+ var downstream = new ExecutionCounterMiddleware();
+ var completion = new CompletionCounterMiddleware();
+ var attempts = 0;
+ var scheduler = terminal
+ .UseMiddleware(new RetryMiddleware(3))
+ .UseMiddleware(downstream)
+ .UseMiddleware(completion);
+
+ var result = await scheduler.Enqueue(() =>
+ {
+ attempts++;
+ if (attempts < 3) throw new InvalidOperationException("retryable");
+ return 42;
+ });
+
+ Assert.That(result, Is.EqualTo(42));
+ Assert.That(terminal.EnqueueCount, Is.EqualTo(1));
+ Assert.That(attempts, Is.EqualTo(3));
+ Assert.That(downstream.CallCount, Is.EqualTo(3));
+ Assert.That(completion.CallCount, Is.EqualTo(1));
+ }
+
+ [Test]
+ public async Task SingleFlight_SharesOneProducerAndAllowsIndependentWaitCancellation()
+ {
+ var terminal = new CountingInlineTerminal();
+ var middleware = new SingleFlightIntMiddleware();
+ var execution = new ExecutionCounterMiddleware();
+ var completion = new CompletionCounterMiddleware();
+ var scheduler = terminal
+ .UseMiddleware(middleware)
+ .UseMiddleware(execution)
+ .UseMiddleware(completion);
+ var producerGate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+ using var firstWait = new CancellationTokenSource();
+
+ var first = scheduler.Enqueue((_, _) => new ValueTask(producerGate.Task), "key", firstWait.Token);
+ await middleware.ProducerStarted;
+ var follower = scheduler.Enqueue((_, _) => throw new AssertionException("A follower must not create another producer."), "key", CancellationToken.None);
+
+ await firstWait.CancelAsync();
+ Assert.That(async () => await first, Throws.InstanceOf());
+ Assert.That(follower.IsCompleted, Is.False);
+
+ producerGate.SetResult(42);
+ Assert.That(await follower, Is.EqualTo(42));
+ Assert.That(terminal.EnqueueCount, Is.EqualTo(1));
+ Assert.That(execution.CallCount, Is.EqualTo(1));
+ Assert.That(completion.CallCount, Is.EqualTo(1));
+ Assert.That(middleware.ActiveCount, Is.Zero);
+ }
+
+ [Test]
+ public async Task DuplicateSuppression_RejectsConcurrentDuplicateAndReleasesKeyAfterCompletion()
+ {
+ var terminal = new CountingInlineTerminal();
+ var middleware = new DuplicateSuppressionMiddleware();
+ var execution = new ExecutionCounterMiddleware();
+ var scheduler = terminal
+ .UseMiddleware(middleware)
+ .UseMiddleware(execution);
+ var gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+
+ var accepted = scheduler.Enqueue((_, _) => new ValueTask(gate.Task), "key", CancellationToken.None);
+ await middleware.Accepted;
+ var duplicate = scheduler.Enqueue((_, _) => new ValueTask(99), "key", CancellationToken.None);
+
+ Assert.That(async () => await duplicate, Throws.TypeOf());
+ Assert.That(terminal.EnqueueCount, Is.EqualTo(1));
+ Assert.That(execution.CallCount, Is.EqualTo(1));
+
+ gate.SetResult(42);
+ Assert.That(await accepted, Is.EqualTo(42));
+ Assert.That(middleware.ActiveCount, Is.Zero);
+
+ Assert.That(await scheduler.Enqueue((_, _) => new ValueTask(7), "key", CancellationToken.None), Is.EqualTo(7));
+ Assert.That(terminal.EnqueueCount, Is.EqualTo(2));
+ Assert.That(execution.CallCount, Is.EqualTo(2));
+ }
+
+ [Test]
+ public async Task KeyedCoordinatorAdapter_EvictsInactiveKeysAndDoesNotTransferOwnership()
+ {
+ var coordinator = new TestKeyedCoordinator();
+ var terminal = new CountingInlineTerminal();
+ var scheduler = terminal.UseMiddleware(new KeyedCoordinatorMiddleware(coordinator));
+ var firstGate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+ var entered = 0;
+
+ var first = scheduler.Enqueue(async (_, _) =>
+ {
+ Interlocked.Increment(ref entered);
+ await firstGate.Task;
+ return 1;
+ }, "same", CancellationToken.None);
+ await coordinator.FirstEntered;
+ var second = scheduler.Enqueue((_, _) =>
+ {
+ Interlocked.Increment(ref entered);
+ return new ValueTask(2);
+ }, "same", CancellationToken.None);
+ var differentKey = scheduler.Enqueue((_, _) => new ValueTask(3), "other", CancellationToken.None);
+
+ await Task.Yield();
+ Assert.That(Volatile.Read(ref entered), Is.EqualTo(1));
+ Assert.That(await differentKey, Is.EqualTo(3), "A different key should not wait for the occupied lane.");
+ firstGate.SetResult(null);
+ var results = await Task.WhenAll(first, second);
+ Assert.That(results[0], Is.EqualTo(1));
+ Assert.That(results[1], Is.EqualTo(2));
+ Assert.That(coordinator.MaximumSameKeyConcurrency, Is.EqualTo(1));
+ Assert.That(coordinator.ActiveKeyCount, Is.Zero);
+ Assert.That(coordinator.IsDisposed, Is.False);
+ Assert.That(terminal.EnqueueCount, Is.EqualTo(3));
+
+ coordinator.Dispose();
+ Assert.That(coordinator.IsDisposed, Is.True);
+ }
+
+ [Test]
+ public async Task LocalScope_TracksSubmissionBeforeMiddlewareAdmissionOnSharedLane()
+ {
+ var lane = new CountingInlineTerminal();
+ var admission = new BlockingAdmissionMiddleware();
+ var decoratedLane = lane.UseMiddleware(admission);
+ var firstScope = new LocalOperationScope(decoratedLane);
+ var secondScope = new LocalOperationScope(decoratedLane);
+
+ var operation = firstScope.Enqueue((_, _) => new ValueTask(42), null, CancellationToken.None);
+ await admission.Entered;
+
+ Assert.That(firstScope.ActiveCount, Is.EqualTo(1));
+ Assert.That(secondScope.ActiveCount, Is.Zero);
+ Assert.That(lane.EnqueueCount, Is.Zero);
+
+ admission.Release();
+ Assert.That(await operation, Is.EqualTo(42));
+ Assert.That(firstScope.ActiveCount, Is.Zero);
+ Assert.That(lane.EnqueueCount, Is.EqualTo(1));
+ }
+
+ [TestCase(OwnershipCancellation.LocalScope)]
+ [TestCase(OwnershipCancellation.SharedLane)]
+ public async Task SharedProducer_DetachesCallerButPreservesOwnershipCancellation(OwnershipCancellation cancellation)
+ {
+ using var localOwner = new CancellationTokenSource();
+ using var sharedLane = new CancellationTokenSource();
+ using var caller = new CancellationTokenSource();
+ var terminal = new CountingInlineTerminal();
+ var ownership = new OwnershipSingleFlightMiddleware(localOwner.Token, sharedLane.Token);
+ var scheduler = terminal
+ .UseMiddleware(ownership)
+ .UseMiddleware(new DropProducerTokenMiddleware());
+ var producerEntered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+
+ var first = scheduler.Enqueue(async (_, producerToken) =>
+ {
+ producerEntered.TrySetResult(null);
+ await Task.Delay(Timeout.InfiniteTimeSpan, producerToken);
+ return 42;
+ }, "key", caller.Token);
+ await producerEntered.Task;
+ var follower = scheduler.Enqueue((_, _) => throw new AssertionException("A follower must not invoke the producer."), "key", CancellationToken.None);
+
+ await caller.CancelAsync();
+ Assert.That(async () => await first, Throws.InstanceOf());
+ Assert.That(follower.IsCompleted, Is.False, "Canceling one caller must not cancel shared producer work.");
+
+ if (cancellation == OwnershipCancellation.LocalScope)
+ {
+ await localOwner.CancelAsync();
+ }
+ else
+ {
+ await sharedLane.CancelAsync();
+ }
+
+ Assert.That(async () => await follower, Throws.InstanceOf());
+ Assert.That(terminal.EnqueueCount, Is.EqualTo(1));
+ Assert.That(() => ownership.ActiveCount, Is.Zero.After(100, 10));
+ }
+
+ private sealed class RetryMiddleware : ITaskSchedulerExecutionMiddleware
+ {
+ private readonly int _maximumAttempts;
+ public RetryMiddleware(int maximumAttempts) => _maximumAttempts = maximumAttempts;
+
+ public async ValueTask InvokeAsync(TaskSchedulerOperationContext context, TaskSchedulerExecutionDelegate continuation)
+ {
+ for (var attempt = 1; ; attempt++)
+ {
+ try
+ {
+ return await continuation(context);
+ }
+ catch (InvalidOperationException) when (attempt < _maximumAttempts)
+ {
+ }
+ }
+ }
+ }
+
+ private sealed class SingleFlightIntMiddleware : ITaskSchedulerEnqueueMiddleware
+ {
+ private readonly ConcurrentDictionary _producers = new ConcurrentDictionary();
+ private readonly TaskCompletionSource _producerStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+
+ public Task ProducerStarted => _producerStarted.Task;
+ public int ActiveCount => _producers.Count;
+
+ public Task InvokeAsync(TaskSchedulerEnqueueContext context, TaskSchedulerEnqueueDelegate continuation)
+ {
+ if (typeof(TResult) != typeof(int)) throw new NotSupportedException("This test prototype supports Int32 results only.");
+ var key = (string)context.State!;
+ var candidate = new Producer();
+ var producer = _producers.GetOrAdd(key, candidate);
+ if (ReferenceEquals(candidate, producer))
+ {
+ _ = Produce(key, producer, context, continuation);
+ }
+
+ return Wait(producer.Task, context.CallerCancellationToken);
+ }
+
+ [SuppressMessage("Design", "CA1031:Do not catch general exception types", Justification = "The single-flight prototype must transfer every producer failure to all follower tasks.")]
+ private async Task Produce(string key, Producer producer, TaskSchedulerEnqueueContext context, TaskSchedulerEnqueueDelegate continuation)
+ {
+ _producerStarted.TrySetResult(null);
+ try
+ {
+ var result = await continuation(context.WithCancellationToken(CancellationToken.None));
+ producer.TrySetResult((int)(object)result!);
+ }
+ catch (Exception exception)
+ {
+ producer.TrySetException(exception);
+ }
+ finally
+ {
+ _producers.TryRemove(new KeyValuePair(key, producer));
+ }
+ }
+
+ private static async Task Wait(Task producer, CancellationToken cancellationToken)
+ => (TResult)(object)await producer.WaitAsync(cancellationToken);
+
+ private sealed class Producer
+ {
+ private readonly TaskCompletionSource _completion = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+ public Task Task => _completion.Task;
+ public void TrySetResult(int result) => _completion.TrySetResult(result);
+ public void TrySetException(Exception exception) => _completion.TrySetException(exception);
+ }
+ }
+
+ private sealed class DuplicateSuppressionMiddleware : ITaskSchedulerEnqueueMiddleware
+ {
+ private readonly ConcurrentDictionary _active = new ConcurrentDictionary();
+ private readonly TaskCompletionSource _accepted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+ public Task Accepted => _accepted.Task;
+ public int ActiveCount => _active.Count;
+
+ public async Task InvokeAsync(TaskSchedulerEnqueueContext context, TaskSchedulerEnqueueDelegate continuation)
+ {
+ var key = (string)context.State!;
+ if (!_active.TryAdd(key, null)) throw new InvalidOperationException("An equivalent operation is already active.");
+ _accepted.TrySetResult(null);
+ try
+ {
+ return await continuation(context);
+ }
+ finally
+ {
+ _active.TryRemove(key, out _);
+ }
+ }
+ }
+
+ private sealed class KeyedCoordinatorMiddleware : ITaskSchedulerEnqueueMiddleware
+ {
+ private readonly TestKeyedCoordinator _coordinator;
+ public KeyedCoordinatorMiddleware(TestKeyedCoordinator coordinator) => _coordinator = coordinator;
+ public Task InvokeAsync(TaskSchedulerEnqueueContext context, TaskSchedulerEnqueueDelegate continuation)
+ => _coordinator.Run((string)context.State!, () => continuation(context));
+ }
+
+ private sealed class TestKeyedCoordinator : IDisposable
+ {
+ private readonly object _sync = new object();
+ private readonly Dictionary _lanes = new Dictionary();
+ private readonly TaskCompletionSource _firstEntered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+ private int _maximumSameKeyConcurrency;
+ private bool _isDisposed;
+
+ public Task FirstEntered => _firstEntered.Task;
+ public int ActiveKeyCount
+ {
+ get
+ {
+ lock (_sync) return _lanes.Count;
+ }
+ }
+ public int MaximumSameKeyConcurrency => _maximumSameKeyConcurrency;
+ public bool IsDisposed => _isDisposed;
+
+ public async Task Run(string key, Func> operation)
+ {
+ Lane lane;
+ lock (_sync)
+ {
+ ObjectDisposedException.ThrowIf(_isDisposed, this);
+ if (!_lanes.TryGetValue(key, out lane!))
+ {
+ lane = new Lane();
+ _lanes.Add(key, lane);
+ }
+
+ lane.Users++;
+ }
+
+ await lane.Gate.WaitAsync();
+ var concurrency = Interlocked.Increment(ref lane.Executing);
+ UpdateMaximum(concurrency);
+ _firstEntered.TrySetResult(null);
+ try
+ {
+ return await operation();
+ }
+ finally
+ {
+ Interlocked.Decrement(ref lane.Executing);
+ lane.Gate.Release();
+ lock (_sync)
+ {
+ lane.Users--;
+ if (lane.Users == 0 && _lanes.TryGetValue(key, out var current) && ReferenceEquals(current, lane))
+ {
+ _lanes.Remove(key);
+ }
+ }
+ }
+ }
+
+ public void Dispose()
+ {
+ lock (_sync) _isDisposed = true;
+ }
+
+ private void UpdateMaximum(int value)
+ {
+ int current;
+ do
+ {
+ current = _maximumSameKeyConcurrency;
+ if (current >= value) return;
+ }
+ while (Interlocked.CompareExchange(ref _maximumSameKeyConcurrency, value, current) != current);
+ }
+
+ private sealed class Lane
+ {
+ public SemaphoreSlim Gate { get; } = new SemaphoreSlim(1, 1);
+ public int Users;
+ public int Executing;
+ }
+ }
+
+ private sealed class LocalOperationScope : ITaskScheduler
+ {
+ private readonly ITaskScheduler _sharedLane;
+ private int _activeCount;
+ public LocalOperationScope(ITaskScheduler sharedLane) => _sharedLane = sharedLane;
+ public int ActiveCount => Volatile.Read(ref _activeCount);
+
+ public async Task Enqueue(Func> taskFunc, object? state, CancellationToken cancellationToken)
+ {
+ Interlocked.Increment(ref _activeCount);
+ try
+ {
+ return await _sharedLane.Enqueue(taskFunc, state, cancellationToken);
+ }
+ finally
+ {
+ Interlocked.Decrement(ref _activeCount);
+ }
+ }
+ }
+
+ private sealed class DropProducerTokenMiddleware : ITaskSchedulerEnqueueMiddleware
+ {
+ public Task InvokeAsync(TaskSchedulerEnqueueContext context, TaskSchedulerEnqueueDelegate continuation)
+ => continuation(context.WithCancellationToken(CancellationToken.None));
+ }
+
+ private sealed class OwnershipSingleFlightMiddleware : ITaskSchedulerEnqueueMiddleware
+ {
+ private readonly CancellationToken _localOwnerToken;
+ private readonly CancellationToken _sharedLaneToken;
+ private readonly ConcurrentDictionary _producers = new ConcurrentDictionary();
+
+ public OwnershipSingleFlightMiddleware(CancellationToken localOwnerToken, CancellationToken sharedLaneToken)
+ {
+ _localOwnerToken = localOwnerToken;
+ _sharedLaneToken = sharedLaneToken;
+ }
+
+ public int ActiveCount => _producers.Count;
+
+ public Task InvokeAsync(TaskSchedulerEnqueueContext context, TaskSchedulerEnqueueDelegate continuation)
+ {
+ if (typeof(TResult) != typeof(int)) throw new NotSupportedException("This test prototype supports Int32 results only.");
+ var key = (string)context.State!;
+ var candidate = new OwnedProducer();
+ var producer = _producers.GetOrAdd(key, candidate);
+ if (ReferenceEquals(candidate, producer))
+ {
+ _ = Produce(key, producer, context, continuation);
+ }
+
+ return Wait(producer.Task, context.CallerCancellationToken, _localOwnerToken, _sharedLaneToken);
+ }
+
+ [SuppressMessage("Design", "CA1031:Do not catch general exception types", Justification = "The ownership prototype must transfer every producer failure to all follower tasks.")]
+ private async Task Produce(string key, OwnedProducer producer, TaskSchedulerEnqueueContext context, TaskSchedulerEnqueueDelegate continuation)
+ {
+ using var ownership = CancellationTokenSource.CreateLinkedTokenSource(context.CancellationToken, _localOwnerToken, _sharedLaneToken);
+ try
+ {
+ var result = await continuation(context.WithCancellationToken(ownership.Token));
+ producer.TrySetResult((int)(object)result!);
+ }
+ catch (Exception exception)
+ {
+ producer.TrySetException(exception);
+ }
+ finally
+ {
+ _producers.TryRemove(new KeyValuePair(key, producer));
+ }
+ }
+
+ private static async Task Wait(Task producer, CancellationToken callerToken, CancellationToken localOwnerToken, CancellationToken sharedLaneToken)
+ {
+ using var wait = CancellationTokenSource.CreateLinkedTokenSource(callerToken, localOwnerToken, sharedLaneToken);
+ return (TResult)(object)await producer.WaitAsync(wait.Token);
+ }
+
+ private sealed class OwnedProducer
+ {
+ private readonly TaskCompletionSource _completion = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+ public Task Task => _completion.Task;
+ public void TrySetResult(int result) => _completion.TrySetResult(result);
+ public void TrySetException(Exception exception) => _completion.TrySetException(exception);
+ }
+ }
+
+ private sealed class BlockingAdmissionMiddleware : ITaskSchedulerEnqueueMiddleware
+ {
+ private readonly TaskCompletionSource _entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+ private readonly TaskCompletionSource _release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+ public Task Entered => _entered.Task;
+ public void Release() => _release.TrySetResult(null);
+
+ public async Task InvokeAsync(TaskSchedulerEnqueueContext context, TaskSchedulerEnqueueDelegate continuation)
+ {
+ _entered.TrySetResult(null);
+ await _release.Task;
+ return await continuation(context);
+ }
+ }
+
+ private sealed class ExecutionCounterMiddleware : ITaskSchedulerExecutionMiddleware
+ {
+ public int CallCount { get; private set; }
+ public ValueTask InvokeAsync(TaskSchedulerOperationContext context, TaskSchedulerExecutionDelegate continuation)
+ {
+ CallCount++;
+ return continuation(context);
+ }
+ }
+
+ private sealed class CompletionCounterMiddleware : ITaskSchedulerCompletionMiddleware
+ {
+ public int CallCount { get; private set; }
+ public ValueTask> InvokeAsync(TaskSchedulerOperationContext context, TaskSchedulerOperationOutcome outcome, TaskSchedulerCompletionDelegate continuation)
+ {
+ CallCount++;
+ return continuation(context, outcome);
+ }
+ }
+
+ private sealed class CountingInlineTerminal : ITaskScheduler
+ {
+ public int EnqueueCount { get; private set; }
+ public async Task Enqueue(Func> taskFunc, object? state, CancellationToken cancellationToken)
+ {
+ EnqueueCount++;
+ return await taskFunc(state, cancellationToken);
+ }
+ }
+
+ }
+}
diff --git a/TaskFlow.Tests/Extensions/InterceptionTaskSchedulerExtensionsFixture.cs b/TaskFlow.Tests/Extensions/InterceptionTaskSchedulerExtensionsFixture.cs
index dccdba2..369c773 100644
--- a/TaskFlow.Tests/Extensions/InterceptionTaskSchedulerExtensionsFixture.cs
+++ b/TaskFlow.Tests/Extensions/InterceptionTaskSchedulerExtensionsFixture.cs
@@ -46,7 +46,7 @@ public void Intercept_ObservesErrorsAndMetadata(ITaskFlow taskFlow)
var interceptor = new RecordingInterceptor(events, false);
var state = new object();
Func> operation = (_, _) => throw new InvalidOperationException("boom");
- var task = taskFlow.Intercept(interceptor).WithOperationName("failure")
+ var task = taskFlow.WithOperationName("failure").Intercept(interceptor)
.Enqueue(operation, state, CancellationToken.None);
Assert.That(async () => await task, Throws.InvalidOperationException.With.Message.EqualTo("boom"));
Assert.That(events, Is.EqualTo(FailedLifecycle));
@@ -174,8 +174,8 @@ public async Task Intercept_SynchronousHooksRunInOrderAndObserveContext(ITaskFlo
var state = new object();
using var cts = new CancellationTokenSource();
- var result = await taskFlow.Intercept(new SynchronousLifecycleInterceptor(recorder))
- .WithOperationName("sync")
+ var result = await taskFlow.WithOperationName("sync")
+ .Intercept(new SynchronousLifecycleInterceptor(recorder))
.Enqueue((operationState, token) =>
{
recorder.Events.Add("operation");
diff --git a/TaskFlow.Tests/Extensions/TaskSchedulerMiddlewareExtensionsFixture.cs b/TaskFlow.Tests/Extensions/TaskSchedulerMiddlewareExtensionsFixture.cs
new file mode 100644
index 0000000..a8b2166
--- /dev/null
+++ b/TaskFlow.Tests/Extensions/TaskSchedulerMiddlewareExtensionsFixture.cs
@@ -0,0 +1,421 @@
+namespace TaskFlow.Tests.Extensions
+{
+ using NUnit.Framework;
+ using System.Collections.Concurrent;
+ using System.Threading.Tasks.Flow;
+
+ [TestFixture]
+ internal sealed class TaskSchedulerMiddlewareExtensionsFixture
+ {
+ private static readonly string[] ExpectedPhaseEvents =
+ {
+ "enqueue:second", "enqueue:first", "terminal", "execute:first", "execute:second",
+ "operation", "complete:first", "complete:second", "terminal-return"
+ };
+
+ [Test]
+ public async Task Pipeline_UsesPhaseOrderingAndOneTerminalDelegate()
+ {
+ var events = new List();
+ var terminal = new RecordingTerminal(events);
+ var scheduler = terminal
+ .UseMiddleware(new RecordingMiddleware("first", events))
+ .UseMiddleware(new RecordingMiddleware("second", events));
+
+ Assert.That(await scheduler.Enqueue(() => { events.Add("operation"); return 42; }), Is.EqualTo(42));
+ Assert.That(terminal.EnqueueCount, Is.EqualTo(1));
+ Assert.That(events, Is.EqualTo(ExpectedPhaseEvents));
+ }
+
+ [Test]
+ public async Task Metadata_IsForwardScopedShadowedAndBranchIndependent()
+ {
+ var observations = new ConcurrentBag();
+ var root = new InlineTerminal().WithAnnotation(new TestAnnotation("root"));
+ var parent = root.UseMiddleware(new AnnotationMiddleware("parent", observations));
+ var left = parent.WithAnnotation(new TestAnnotation("left"))
+ .UseMiddleware(new AnnotationMiddleware("left", observations));
+ var right = parent.WithAnnotation(new TestAnnotation("right"))
+ .UseMiddleware(new AnnotationMiddleware("right", observations));
+
+ await Task.WhenAll(left.Enqueue(() => 1), right.Enqueue(() => 2));
+
+ Assert.That(observations.Count(x => x == "parent:root"), Is.EqualTo(2));
+ Assert.That(observations, Does.Contain("left:left"));
+ Assert.That(observations, Does.Contain("right:right"));
+ }
+
+ [Test]
+ public async Task EnqueueMiddleware_MayShortCircuitWithoutReachingTerminal()
+ {
+ var terminal = new RecordingTerminal(new List());
+ var scheduler = terminal.UseMiddleware(new ConstantMiddleware(17));
+
+ Assert.That(await scheduler.Enqueue(() => 42), Is.EqualTo(17));
+ Assert.That(terminal.EnqueueCount, Is.Zero);
+ }
+
+ [Test]
+ public async Task ExecutionMiddleware_MayInvokeContinuationMoreThanOnceInOneQueueTurn()
+ {
+ var terminal = new RecordingTerminal(new List());
+ var calls = 0;
+ var scheduler = terminal.UseMiddleware(new TwiceMiddleware());
+
+ Assert.That(await scheduler.Enqueue(() => ++calls), Is.EqualTo(2));
+ Assert.That(calls, Is.EqualTo(2));
+ Assert.That(terminal.EnqueueCount, Is.EqualTo(1));
+ }
+
+ [Test]
+ public async Task CompoundMiddleware_SharesOnlyItsRegistrationLocalState()
+ {
+ var middleware = new CompoundStateMiddleware();
+ var scheduler = new InlineTerminal().UseMiddleware(middleware);
+
+ await Task.WhenAll(scheduler.Enqueue(() => 1), scheduler.Enqueue(() => 2));
+
+ Assert.That(middleware.ObservedIds, Has.Count.EqualTo(2));
+ Assert.That(middleware.ObservedIds.Distinct().Count(), Is.EqualTo(2));
+ }
+
+ [Test]
+ public async Task Terminal_ReceivesOriginalStateAndProducerToken()
+ {
+ var terminal = new RecordingTerminal(new List());
+ var state = new object();
+ using var source = new CancellationTokenSource();
+ var scheduler = terminal.UseMiddleware(new AnnotationMiddleware("unused", new ConcurrentBag()));
+
+ await scheduler.Enqueue((s, _) => new ValueTask(s), state, source.Token);
+
+ Assert.That(terminal.State, Is.SameAs(state));
+ Assert.That(terminal.CancellationToken, Is.EqualTo(source.Token));
+ }
+
+ [Test]
+ public async Task EnqueueWithContext_ReceivesFinalStateTokensAndAnnotations()
+ {
+ var state = new object();
+ using var source = new CancellationTokenSource();
+ var scheduler = new InlineTerminal().WithAnnotation(new TestAnnotation("final"));
+
+ var result = await scheduler.EnqueueWithContext(
+ context => new ValueTask<(object?, CancellationToken, CancellationToken, string?)>((
+ context.State,
+ context.CallerCancellationToken,
+ context.CancellationToken,
+ context.GetAnnotation()?.Value)),
+ state,
+ source.Token);
+
+ Assert.That(result.Item1, Is.SameAs(state));
+ Assert.That(result.Item2, Is.EqualTo(source.Token));
+ Assert.That(result.Item3, Is.EqualTo(source.Token));
+ Assert.That(result.Item4, Is.EqualTo("final"));
+ }
+
+ [Test]
+ public async Task EnqueueMiddleware_MayReplaceProducerTokenWithoutReplacingCallerToken()
+ {
+ using var callerSource = new CancellationTokenSource();
+ using var producerSource = new CancellationTokenSource();
+ var terminal = new RecordingTerminal(new List());
+ var observer = new TokenObservingMiddleware();
+ var scheduler = terminal
+ .UseMiddleware(observer)
+ .UseMiddleware(new ProducerTokenMiddleware(producerSource.Token));
+
+ await scheduler.Enqueue(
+ (_, _) => new ValueTask(42),
+ null,
+ callerSource.Token);
+
+ Assert.That(terminal.CancellationToken, Is.EqualTo(producerSource.Token));
+ Assert.That(observer.CallerToken, Is.EqualTo(callerSource.Token));
+ Assert.That(observer.ProducerToken, Is.EqualTo(producerSource.Token));
+ }
+
+ [Test]
+ public void TerminalEnqueueFailure_IsProcessedByCompletionExactlyOnce()
+ {
+ var failure = new InvalidOperationException("rejected");
+ var completion = new ObservingCompletionMiddleware();
+ var scheduler = new ThrowingTerminal(failure).UseMiddleware(completion);
+
+ var thrown = Assert.ThrowsAsync(async () => await scheduler.Enqueue(() => 42));
+
+ Assert.That(thrown, Is.SameAs(failure));
+ Assert.That(completion.CallCount, Is.EqualTo(1));
+ Assert.That(completion.ObservedException, Is.SameAs(failure));
+ }
+
+ [Test]
+ public void ExecutionFailure_PreservesExceptionAndRunsCompletionExactlyOnce()
+ {
+ var failure = CreateFailureWithCapturedStack();
+ var completion = new ObservingCompletionMiddleware();
+ var scheduler = new InlineTerminal().UseMiddleware(completion);
+
+ var thrown = Assert.ThrowsAsync(async () => await scheduler.Enqueue(
+ (_, _) => ValueTask.FromException(failure),
+ null,
+ CancellationToken.None));
+
+ Assert.That(thrown, Is.SameAs(failure));
+ Assert.That(thrown!.StackTrace, Does.Contain(nameof(CreateFailureWithCapturedStack)));
+ Assert.That(completion.CallCount, Is.EqualTo(1));
+ Assert.That(completion.ObservedException, Is.SameAs(failure));
+ }
+
+ [Test]
+ public void CompletionReplacement_IsVisibleToLaterMiddlewareAndCaller()
+ {
+ var original = new InvalidOperationException("original");
+ var replacement = new NotSupportedException("replacement");
+ var observer = new ObservingCompletionMiddleware();
+ var scheduler = new InlineTerminal()
+ .UseMiddleware(new ReplacingCompletionMiddleware(replacement))
+ .UseMiddleware(observer);
+
+ var thrown = Assert.ThrowsAsync(async () => await scheduler.Enqueue(
+ (_, _) => ValueTask.FromException(original),
+ null,
+ CancellationToken.None));
+
+ Assert.That(thrown, Is.SameAs(replacement));
+ Assert.That(observer.ObservedException, Is.SameAs(replacement));
+ }
+
+ [Test]
+ public void CompletionFailureBeforeContinuation_IsVisibleToLaterMiddleware()
+ {
+ var replacement = new NotSupportedException("callback failed");
+ var observer = new ObservingCompletionMiddleware();
+ var scheduler = new InlineTerminal()
+ .UseMiddleware(new ThrowingCompletionMiddleware(replacement, throwAfterContinuation: false))
+ .UseMiddleware(observer);
+
+ var thrown = Assert.ThrowsAsync(async () => await scheduler.Enqueue(() => 42));
+
+ Assert.That(thrown, Is.SameAs(replacement));
+ Assert.That(observer.CallCount, Is.EqualTo(1));
+ Assert.That(observer.ObservedException, Is.SameAs(replacement));
+ }
+
+ [Test]
+ public void CompletionFailureAfterContinuation_ReplacesFinalOutcomeWithoutRepeatingLaterMiddleware()
+ {
+ var replacement = new NotSupportedException("callback failed after next");
+ var observer = new ObservingCompletionMiddleware();
+ var scheduler = new InlineTerminal()
+ .UseMiddleware(new ThrowingCompletionMiddleware(replacement, throwAfterContinuation: true))
+ .UseMiddleware(observer);
+
+ var thrown = Assert.ThrowsAsync(async () => await scheduler.Enqueue(() => 42));
+
+ Assert.That(thrown, Is.SameAs(replacement));
+ Assert.That(observer.CallCount, Is.EqualTo(1));
+ Assert.That(observer.ObservedException, Is.Null);
+ }
+
+ [Test]
+ public void UseMiddleware_RejectsMarkerWithoutPhase()
+ {
+ Assert.That(
+ () => new InlineTerminal().UseMiddleware(new MarkerOnlyMiddleware()),
+ Throws.ArgumentException.With.Property("ParamName").EqualTo("middleware"));
+ }
+
+ private static InvalidOperationException CreateFailureWithCapturedStack()
+ {
+ try
+ {
+ throw new InvalidOperationException("failure");
+ }
+ catch (InvalidOperationException exception)
+ {
+ return exception;
+ }
+ }
+
+ private sealed class TestAnnotation : IOperationAnnotation
+ {
+ public TestAnnotation(string value) => Value = value;
+ public string Value { get; }
+ }
+
+ private sealed class RecordingMiddleware : ITaskSchedulerEnqueueMiddleware, ITaskSchedulerExecutionMiddleware, ITaskSchedulerCompletionMiddleware
+ {
+ private readonly string _name;
+ private readonly IList _events;
+ public RecordingMiddleware(string name, IList events) { _name = name; _events = events; }
+
+ public async Task InvokeAsync(TaskSchedulerEnqueueContext context, TaskSchedulerEnqueueDelegate continuation)
+ {
+ _events.Add("enqueue:" + _name);
+ return await continuation(context);
+ }
+
+ public async ValueTask InvokeAsync(TaskSchedulerOperationContext context, TaskSchedulerExecutionDelegate continuation)
+ {
+ _events.Add("execute:" + _name);
+ return await continuation(context);
+ }
+
+ public async ValueTask> InvokeAsync(TaskSchedulerOperationContext context, TaskSchedulerOperationOutcome outcome, TaskSchedulerCompletionDelegate continuation)
+ {
+ _events.Add("complete:" + _name);
+ return await continuation(context, outcome);
+ }
+ }
+
+ private sealed class AnnotationMiddleware : ITaskSchedulerExecutionMiddleware
+ {
+ private readonly string _name;
+ private readonly ConcurrentBag _observations;
+ public AnnotationMiddleware(string name, ConcurrentBag observations) { _name = name; _observations = observations; }
+ public ValueTask InvokeAsync(TaskSchedulerOperationContext context, TaskSchedulerExecutionDelegate continuation)
+ {
+ _observations.Add(_name + ":" + context.GetAnnotation()?.Value);
+ return continuation(context);
+ }
+ }
+
+ private sealed class ConstantMiddleware : ITaskSchedulerEnqueueMiddleware
+ {
+ private readonly int _value;
+ public ConstantMiddleware(int value) => _value = value;
+ public Task InvokeAsync(TaskSchedulerEnqueueContext context, TaskSchedulerEnqueueDelegate continuation)
+ => Task.FromResult((TResult)(object)_value);
+ }
+
+ private sealed class ProducerTokenMiddleware : ITaskSchedulerEnqueueMiddleware
+ {
+ private readonly CancellationToken _producerToken;
+ public ProducerTokenMiddleware(CancellationToken producerToken) => _producerToken = producerToken;
+ public Task InvokeAsync(TaskSchedulerEnqueueContext context, TaskSchedulerEnqueueDelegate continuation)
+ => continuation(context.WithCancellationToken(_producerToken));
+ }
+
+ private sealed class TokenObservingMiddleware : ITaskSchedulerExecutionMiddleware
+ {
+ public CancellationToken CallerToken { get; private set; }
+ public CancellationToken ProducerToken { get; private set; }
+ public ValueTask InvokeAsync(TaskSchedulerOperationContext context, TaskSchedulerExecutionDelegate continuation)
+ {
+ CallerToken = context.CallerCancellationToken;
+ ProducerToken = context.CancellationToken;
+ return continuation(context);
+ }
+ }
+
+ private sealed class ObservingCompletionMiddleware : ITaskSchedulerCompletionMiddleware
+ {
+ public int CallCount { get; private set; }
+ public Exception? ObservedException { get; private set; }
+ public ValueTask> InvokeAsync(TaskSchedulerOperationContext context, TaskSchedulerOperationOutcome outcome, TaskSchedulerCompletionDelegate continuation)
+ {
+ CallCount++;
+ ObservedException = outcome.Exception;
+ return continuation(context, outcome);
+ }
+ }
+
+ private sealed class ReplacingCompletionMiddleware : ITaskSchedulerCompletionMiddleware
+ {
+ private readonly Exception _replacement;
+ public ReplacingCompletionMiddleware(Exception replacement) => _replacement = replacement;
+ public ValueTask> InvokeAsync(TaskSchedulerOperationContext context, TaskSchedulerOperationOutcome outcome, TaskSchedulerCompletionDelegate continuation)
+ => continuation(context, TaskSchedulerOperationOutcome.FromException(_replacement));
+ }
+
+ private sealed class ThrowingCompletionMiddleware : ITaskSchedulerCompletionMiddleware
+ {
+ private readonly Exception _exception;
+ private readonly bool _throwAfterContinuation;
+ public ThrowingCompletionMiddleware(Exception exception, bool throwAfterContinuation)
+ {
+ _exception = exception;
+ _throwAfterContinuation = throwAfterContinuation;
+ }
+
+ public async ValueTask> InvokeAsync(TaskSchedulerOperationContext context, TaskSchedulerOperationOutcome outcome, TaskSchedulerCompletionDelegate continuation)
+ {
+ if (!_throwAfterContinuation) throw _exception;
+ _ = await continuation(context, outcome);
+ throw _exception;
+ }
+ }
+
+ private sealed class MarkerOnlyMiddleware : ITaskSchedulerMiddleware
+ {
+ }
+
+ private sealed class TwiceMiddleware : ITaskSchedulerExecutionMiddleware
+ {
+ public async ValueTask InvokeAsync(TaskSchedulerOperationContext context, TaskSchedulerExecutionDelegate continuation)
+ {
+ _ = await continuation(context);
+ return await continuation(context);
+ }
+ }
+
+ private sealed class CompoundStateMiddleware : ITaskSchedulerEnqueueMiddleware, ITaskSchedulerExecutionMiddleware
+ {
+ private int _nextId;
+ public ConcurrentBag ObservedIds { get; } = new ConcurrentBag();
+ public Task InvokeAsync(TaskSchedulerEnqueueContext context, TaskSchedulerEnqueueDelegate continuation)
+ {
+ context.GetOrCreateLocalState(() => new LocalState(Interlocked.Increment(ref _nextId)));
+ return continuation(context);
+ }
+
+ public ValueTask InvokeAsync(TaskSchedulerOperationContext context, TaskSchedulerExecutionDelegate continuation)
+ {
+ ObservedIds.Add(context.GetLocalState()!.Id);
+ return continuation(context);
+ }
+ }
+
+ private sealed class LocalState
+ {
+ public LocalState(int id) => Id = id;
+ public int Id { get; }
+ }
+
+ private sealed class RecordingTerminal : ITaskScheduler
+ {
+ private readonly IList _events;
+ public RecordingTerminal(IList events) => _events = events;
+ public int EnqueueCount { get; private set; }
+ public object? State { get; private set; }
+ public CancellationToken CancellationToken { get; private set; }
+
+ public async Task Enqueue(Func> taskFunc, object? state, CancellationToken cancellationToken)
+ {
+ EnqueueCount++;
+ State = state;
+ CancellationToken = cancellationToken;
+ _events.Add("terminal");
+ var result = await taskFunc(state, cancellationToken);
+ _events.Add("terminal-return");
+ return result;
+ }
+ }
+
+ private sealed class InlineTerminal : ITaskScheduler
+ {
+ public async Task Enqueue(Func> taskFunc, object? state, CancellationToken cancellationToken)
+ => await taskFunc(state, cancellationToken);
+ }
+
+ private sealed class ThrowingTerminal : ITaskScheduler
+ {
+ private readonly Exception _exception;
+ public ThrowingTerminal(Exception exception) => _exception = exception;
+ public Task Enqueue(Func> taskFunc, object? state, CancellationToken cancellationToken)
+ => Task.FromException(_exception);
+ }
+ }
+}
diff --git a/TaskFlow.Tests/Extensions/TimeoutTaskSchedulerExtensionsFixture.cs b/TaskFlow.Tests/Extensions/TimeoutTaskSchedulerExtensionsFixture.cs
index 7f90066..120582c 100644
--- a/TaskFlow.Tests/Extensions/TimeoutTaskSchedulerExtensionsFixture.cs
+++ b/TaskFlow.Tests/Extensions/TimeoutTaskSchedulerExtensionsFixture.cs
@@ -75,8 +75,8 @@ public void Timeout_WhenOperationNameSpecified_ShouldThrowTimeoutExceptionWithOp
_taskFlow = taskFlow;
var task = taskFlow
- .WithTimeout(TimeSpan.FromMilliseconds(100))
.WithOperationName("inner")
+ .WithTimeout(TimeSpan.FromMilliseconds(100))
.CreateCancelPrevious()
.WithOperationName("outer")
.Enqueue(
@@ -88,7 +88,7 @@ public void Timeout_WhenOperationNameSpecified_ShouldThrowTimeoutExceptionWithOp
42,
CancellationToken.None);
- Assert.That(async () => await task.ConfigureAwait(false), Throws.InstanceOf().With.Message.Contain("outer"));
+ Assert.That(async () => await task.ConfigureAwait(false), Throws.InstanceOf().With.Message.Contain("inner"));
}
[TestCaseSource(typeof(TaskFlows), nameof(TaskFlows.CreateTaskFlows))]
diff --git a/TaskFlow.Tests/Extensions/WrapperOwnershipTaskSchedulerExtensionsFixture.cs b/TaskFlow.Tests/Extensions/WrapperOwnershipTaskSchedulerExtensionsFixture.cs
index 3bfe54c..ab0e070 100644
--- a/TaskFlow.Tests/Extensions/WrapperOwnershipTaskSchedulerExtensionsFixture.cs
+++ b/TaskFlow.Tests/Extensions/WrapperOwnershipTaskSchedulerExtensionsFixture.cs
@@ -24,6 +24,8 @@ public void ExtensionWrappers_ShouldNotOwnTaskFlow_DisposableInterfacesAreNotExp
var cancelPrevious = taskFlow.CreateCancelPrevious();
var cancellationScope = taskFlow.CreateCancellationScope(CancellationToken.None);
var intercepted = taskFlow.Intercept(new NoOpAsyncInterceptorFactory());
+ var annotated = taskFlow.WithOperationName("non-owning");
+ var middleware = taskFlow.UseMiddleware(new NoOpMiddleware());
Assert.That(timeout, Is.Not.InstanceOf());
Assert.That(timeout, Is.Not.InstanceOf());
@@ -39,6 +41,12 @@ public void ExtensionWrappers_ShouldNotOwnTaskFlow_DisposableInterfacesAreNotExp
Assert.That(intercepted, Is.Not.InstanceOf());
Assert.That(intercepted, Is.Not.InstanceOf());
+
+ Assert.That(annotated, Is.Not.InstanceOf());
+ Assert.That(annotated, Is.Not.InstanceOf());
+
+ Assert.That(middleware, Is.Not.InstanceOf());
+ Assert.That(middleware, Is.Not.InstanceOf());
}
private sealed class NoOpAsyncInterceptorFactory : IAsyncTaskSchedulerInterceptor
@@ -71,5 +79,13 @@ public ValueTask OnSuccessAsync(TaskSchedulerInterceptionContext contex
return default;
}
}
+
+ private sealed class NoOpMiddleware : ITaskSchedulerExecutionMiddleware
+ {
+ public ValueTask InvokeAsync(TaskSchedulerOperationContext context, TaskSchedulerExecutionDelegate continuation)
+ {
+ return continuation(context);
+ }
+ }
}
}
diff --git a/TaskFlow.sln b/TaskFlow.sln
index 4586601..d4e8130 100644
--- a/TaskFlow.sln
+++ b/TaskFlow.sln
@@ -1,7 +1,7 @@
Microsoft Visual Studio Solution File, Format Version 12.00
-# Visual Studio Version 17
-VisualStudioVersion = 17.3.32901.215
+# Visual Studio Version 18
+VisualStudioVersion = 18.9.12105.275 stable
MinimumVisualStudioVersion = 10.0.40219.1
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "TaskFlow", "TaskFlow\TaskFlow.csproj", "{7B846A17-1146-4BF1-9E01-81AAFB097A02}"
EndProject
@@ -68,6 +68,7 @@ Global
GlobalSection(SharedMSBuildProjectFiles) = preSolution
TaskFlow.Annotations\TaskFlow.Annotations.projitems*{38b4a885-e50e-4339-91f6-b07976dce941}*SharedItemsImports = 5
TaskFlow.Annotations\TaskFlow.Annotations.projitems*{7b846a17-1146-4bf1-9e01-81aafb097a02}*SharedItemsImports = 5
+ TaskFlow.Annotations\TaskFlow.Annotations.projitems*{a8d73e43-4792-4a8d-a57a-2df8e91ccf88}*SharedItemsImports = 5
TaskFlow.Annotations\TaskFlow.Annotations.projitems*{b9a7fa49-0771-408d-82cf-42ea7ebe0f31}*SharedItemsImports = 13
TaskFlow.Annotations\TaskFlow.Annotations.projitems*{dd1d5309-b058-4563-8bb3-96b93ab7731f}*SharedItemsImports = 5
EndGlobalSection
diff --git a/TaskFlow/Extensions/AnnotatingTaskSchedulerExtensions.cs b/TaskFlow/Extensions/AnnotatingTaskSchedulerExtensions.cs
index f920562..b250628 100644
--- a/TaskFlow/Extensions/AnnotatingTaskSchedulerExtensions.cs
+++ b/TaskFlow/Extensions/AnnotatingTaskSchedulerExtensions.cs
@@ -111,7 +111,7 @@ public static ITaskScheduler WithOperationName(this ITaskScheduler taskScheduler
{
Argument.NotEmpty(operationName);
- return taskScheduler.WithExtendedState(new OperationNameAnnotation( operationName));
+ return taskScheduler.WithAnnotation(new OperationNameAnnotation(operationName));
}
///
@@ -180,11 +180,10 @@ public static Task AnnotatedEnqueue(this ITaskScheduler taskS
Argument.NotNull(taskScheduler);
Argument.NotNull(taskFunc);
- var annotation = (state as ExtendedState)
- .Unwrap()
- .FirstOrDefault();
-
- return taskScheduler.Enqueue((s, c) => taskFunc(s, annotation, c), state, cancellationToken);
+ return taskScheduler.EnqueueWithContext(
+ context => taskFunc(context.State, context.GetAnnotation(), context.CancellationToken),
+ state,
+ cancellationToken);
}
}
-}
\ No newline at end of file
+}
diff --git a/TaskFlow/Extensions/CancelPreviousTaskSchedulerExtensions.cs b/TaskFlow/Extensions/CancelPreviousTaskSchedulerExtensions.cs
index 68a326f..7a2880f 100644
--- a/TaskFlow/Extensions/CancelPreviousTaskSchedulerExtensions.cs
+++ b/TaskFlow/Extensions/CancelPreviousTaskSchedulerExtensions.cs
@@ -183,30 +183,27 @@ public static class CancelPreviousTaskSchedulerExtensions
///
public static ITaskScheduler CreateCancelPrevious(this ITaskScheduler taskScheduler)
{
- return new CancelPreviousTaskSchedulerWrapper(taskScheduler);
+ Argument.NotNull(taskScheduler);
+ return taskScheduler.UseMiddleware(new CancelPreviousMiddleware());
}
- private sealed class CancelPreviousTaskSchedulerWrapper : ITaskScheduler
+ private sealed class CancelPreviousMiddleware : ITaskSchedulerEnqueueMiddleware
{
- private readonly ITaskScheduler _baseTaskScheduler;
private readonly CancelAllTokensAllocator _cancelAllTokensAllocator;
- public CancelPreviousTaskSchedulerWrapper(ITaskScheduler baseTaskScheduler)
+ public CancelPreviousMiddleware()
{
- Argument.NotNull(baseTaskScheduler);
-
- _baseTaskScheduler = baseTaskScheduler;
_cancelAllTokensAllocator = new CancelAllTokensAllocator();
}
- public async Task Enqueue(Func> taskFunc, object? state, CancellationToken cancellationToken)
+ public async Task