From 97f240b34032e32b90f54b90c3baae0f42a2375f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ricardo=20Gro=C3=9F?= Date: Thu, 29 Oct 2026 14:11:04 +0100 Subject: [PATCH] feat: Choose, SequenceAsync and TraverseAsync read an async stream Choose() and Choose(selector) over an IAsyncEnumerable return one, lazily, the caller's token reaching the source through [EnumeratorCancellation]. SequenceAsync(cancellationToken) over a stream of either Result shape, and the four TraverseAsync overloads over a stream, read the next element only after the one before succeeded, check a token they are given before each read and each operation, and dispose the enumerator on every exit. Measured before the code (design/option-result.md): no name binds ambiguously beside .NET 10's System.Linq.AsyncEnumerable, and a T[] or an EF Core 10 DbSet still binds to the sequence members; a type that is both IEnumerable and IAsyncEnumerable is ambiguous for Choose(selector) and TraverseAsync as it is for the BCL's own Where. CMTK0003 reports a dropped SequenceAsync or stream TraverseAsync by its type with no analyzer change; the new analyzer test pins it. Variants: each behaviour runs over a stream whose elements arrive at once and one whose elements each wait, nullable and non-nullable elements, both Result shapes, both token forms. The completeness tests (NullDelegateTests, UninitializedResultTests, AsynchronousContextTests, the last widened to members returning an async stream) took the new members. Thirteen seeded defects in src/ (a token not passed, a check dropped, ConfigureAwait(false) dropped, an enumerator left undisposed, a selector checked late) each fail a test; the stream TraverseAsync's context case first missed one and now reads a pending stream. Closes #128 Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 1 + design/option-result.md | 14 +- docs/option.md | 1 + docs/result.md | 2 + src/CodoMetis.TypeKit/OptionExtensions.cs | 27 ++ src/CodoMetis.TypeKit/README.md | 15 + src/CodoMetis.TypeKit/ResultExtensions.cs | 192 ++++++++ .../SequenceOptionExtensions.cs | 24 +- .../IgnoredOutcomeAnalyzerTests.cs | 44 ++ .../Base/ResultSamples.cs | 15 + .../AsyncStreamTests.cs | 445 ++++++++++++++++++ .../AsynchronousContextTests.cs | 44 +- .../NullDelegateTests.cs | 10 + test/CodoMetis.TypeKit.Tests/NullTaskTests.cs | 6 +- .../UninitializedResultTests.cs | 6 + test/oracles/claims.json | 72 +++ test/oracles/surface-waivers.json | 1 - 17 files changed, 910 insertions(+), 9 deletions(-) create mode 100644 test/CodoMetis.TypeKit.Tests/AsyncStreamTests.cs diff --git a/CHANGELOG.md b/CHANGELOG.md index d33373ac..bc1b2d01 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -17,6 +17,7 @@ section is the GitHub release's notes. - AspNetCore: `FaultResponse.WithoutFaultName()` answers a member without naming it: no `fault`, no `rule` and no member description in the problem, so two causes on one status look alike; the document says that status names no fault, and `IFaultFeature.NameWithheld` tells the host ([Faults a client must not tell apart](https://github.com/CaffeinatedCoder/CodoMetis.TypeKit/blob/main/src/CodoMetis.TypeKit.AspNetCore/README.md#faults-a-client-must-not-tell-apart)). - `Result.CatchAsync` takes a call that returns a `ValueTask`, `stream.ReadAsync(buffer)` without `.AsTask()`; an `async` lambda binds as before ([An exception as an outcome](https://github.com/CaffeinatedCoder/CodoMetis.TypeKit/blob/main/docs/result.md#an-exception-as-an-outcome)). - `TraverseAsync` on `Option` and `Patch`: an asynchronous check of an optional input or a patch field, with a token form; nothing is called without a value ([Collect many results](https://github.com/CaffeinatedCoder/CodoMetis.TypeKit/blob/main/docs/result.md#collect-many-results)). +- `Choose()`, `Choose(selector)`, `SequenceAsync` and `TraverseAsync` read an `IAsyncEnumerable`, in order and without a list first; the token reaches the stream, nothing is read after the first error, and the enumerator is disposed however they stop ([Chain steps](https://github.com/CaffeinatedCoder/CodoMetis.TypeKit/blob/main/src/CodoMetis.TypeKit/README.md#chain-steps)). - `Fault.Cause`: `Fault.Of(reason, detail, exception)` keeps the exception underneath for your log; equality and `ToString()` leave it out, and `ToHttpResult()` hands it to the host as `IFaultFeature.Cause` and `FaultProblem.Cause`, never into the problem ([What failed underneath](https://github.com/CaffeinatedCoder/CodoMetis.TypeKit/blob/main/src/CodoMetis.TypeKit.AspNetCore/README.md#what-failed-underneath)). - Testing: every assertion works on a `Task` or `ValueTask` of what it checks, `await service.FindAsync(id).ShouldBeSome()`; a task that throws fails the test with its own exception, and an assertion left without `await` is reported (CMTK0003) ([CodoMetis.TypeKit.Testing](https://github.com/CaffeinatedCoder/CodoMetis.TypeKit/blob/main/src/CodoMetis.TypeKit.Testing/README.md)). - Analyzers: a validated value object without `Create` gets a code fix, **Add Create**, that writes a `Create` accepting every value, ready for your rules ([Add Create](https://github.com/CaffeinatedCoder/CodoMetis.TypeKit/blob/main/src/CodoMetis.TypeKit.Analyzers/README.md#add-create)). diff --git a/design/option-result.md b/design/option-result.md index 17a25a21..07930acb 100644 --- a/design/option-result.md +++ b/design/option-result.md @@ -86,7 +86,19 @@ is the value-less shape. Anything that turns this into a second LanguageExt is o once on it, so in a UI or a Blazor component a callback that touches dispatcher-bound state works or fails by timing. That is documented rather than changed: await the task and call the synchronous combinator, or marshal inside the callback. `AsynchronousContextTests` holds every public member - that returns a task to it. + that returns a task or an async stream to it. +- **Async streams (2.3.0, #128).** `Choose()` and `Choose(selector)` over an `IAsyncEnumerable` + return one, lazily, the caller's token reaching the source through `[EnumeratorCancellation]`. + `SequenceAsync(cancellationToken)` and `TraverseAsync` over one read the next element only after the + one before it succeeded, check a token they are given before each read and each operation, and + dispose the enumerator on every exit. Measured 2026-10-29 against .NET 10's + `System.Linq.AsyncEnumerable` and EF Core 10.0.12: no name binds ambiguously beside its operators, + and a `T[]` or a `DbSet` (which EF Core 10 does not convert to `IAsyncEnumerable`) still binds + to the sequence members. A type that implements both `IEnumerable` and `IAsyncEnumerable` is + ambiguous (CS0121) for `Choose(selector)` and `TraverseAsync`, as it is for the BCL's own `Where`. That + is left as the BCL leaves it, not settled by an overload priority: a silent pick of either reading is + worse than a compile error. CMTK0003 reports a dropped `SequenceAsync` by its type, as it does + `TraverseAsync`, with no change to the analyzer. **Decided 2026-09-27: an uninitialized `Result` throws.** - A `default` result is `State == Uninitialized`. If the branching members tested only diff --git a/docs/option.md b/docs/option.md index 17b08be5..d3ea841e 100644 --- a/docs/option.md +++ b/docs/option.md @@ -189,6 +189,7 @@ elements `Choose(selector)` reads may be null, and only what the selector finds | `GetValueOrNone(key)` | `IReadOnlyDictionary` | `Option` | | `Choose(selector)` | `IEnumerable`, its elements nullable or not | the values the selector answers `Some` for, in order and lazily: a filter and a map in one | | `Choose()` | `IEnumerable>` | the values of the options that hold one, in order and lazily | +| `Choose(selector)`, `Choose()` | `IAsyncEnumerable`, `IAsyncEnumerable>` | the same, as an `IAsyncEnumerable` of the values as they arrive; the token given to `WithCancellation` reaches the stream | | `IsSome()`, `IsNone()` | `Option` | `bool` | | `Match(onSome, onNone)` | `Option` | what the branch returns | | `TryGetValue(out value)` | `Option` | `bool`; `value` is non-null when true | diff --git a/docs/result.md b/docs/result.md index 66f038c3..0ba9d4f0 100644 --- a/docs/result.md +++ b/docs/result.md @@ -323,9 +323,11 @@ throughout. | `Traverse(selector)` | `Option` | `Result, TError>`: an optional input validated when present | | `Traverse(selector)` | `Patch` | `Result, TError>`: a patch field validated when it carries a value; a clear or an absence passes through | | `TraverseAsync(selector)`, `TraverseAsync(selector, cancellationToken)` | `IEnumerable` | as `Traverse`, one operation awaited at a time; nothing starts after an error, or once the token is cancelled | +| `TraverseAsync(selector)`, `TraverseAsync(selector, cancellationToken)` | `IAsyncEnumerable` | as over `IEnumerable`; nothing more is read after an error, and the token reaches the stream too | | `TraverseAsync(selector)`, `TraverseAsync(selector, cancellationToken)` | `Option`, `Patch` | as `Traverse`, the operation awaited; nothing is called without a value | | `Transpose()` | `Option>`, `Result, TError>` | the one turned inside out: `None` is a success without a value, an error stays the error | | `Sequence()` | `IEnumerable>`, `IEnumerable>` | the list of values, or the first error | +| `SequenceAsync()`, `SequenceAsync(cancellationToken)` | `IAsyncEnumerable>`, `IAsyncEnumerable>` | the same, awaited; nothing more is read after the first error, and the token reaches the stream | **Reaching the outcome** diff --git a/src/CodoMetis.TypeKit/OptionExtensions.cs b/src/CodoMetis.TypeKit/OptionExtensions.cs index 369d57a6..6895c310 100644 --- a/src/CodoMetis.TypeKit/OptionExtensions.cs +++ b/src/CodoMetis.TypeKit/OptionExtensions.cs @@ -651,6 +651,24 @@ public IEnumerable Choose() } } + /// The stream of options. + /// The type of the values. + extension(IAsyncEnumerable> options) where T : notnull + { + /// + /// The values of the options that hold one, in order, as they arrive: a stream of lookups to what + /// was found. Lazy, as an async LINQ operator is, and the token given to WithCancellation or + /// GetAsyncEnumerator reaches the stream it reads. + /// + /// The values, in order. + public IAsyncEnumerable Choose() + { + ArgumentNullException.ThrowIfNull(options); + + return Chosen(options, static option => option); + } + } + /// The nullable value. /// The underlying value type. extension(in T? instance) where T : struct @@ -679,4 +697,13 @@ internal static IEnumerable Chosen(IEnumerable source, F if (selector(element).TryGetValue(out var value)) yield return value; } } + + /// The iterator behind both Choose over a stream; the caller's token is the stream's. + internal static async IAsyncEnumerable Chosen(IAsyncEnumerable source, Func> selector, [EnumeratorCancellation] CancellationToken cancellationToken = default) where TResult : notnull + { + await foreach (var element in source.WithCancellation(cancellationToken).ConfigureAwait(false)) + { + if (selector(element).TryGetValue(out var value)) yield return value; + } + } } diff --git a/src/CodoMetis.TypeKit/README.md b/src/CodoMetis.TypeKit/README.md index 2c52d208..37d27f6c 100644 --- a/src/CodoMetis.TypeKit/README.md +++ b/src/CodoMetis.TypeKit/README.md @@ -79,6 +79,7 @@ Option discount = // a later from - **`Choose(selector)`** keeps the values a selector returns `Some` for, in order and without the `None`s. It filters and maps in one step, over elements that may be null (`nicknames.Choose(n => n.ToOption())`). `Choose()` keeps the values a sequence of options holds. + Both read an async stream too, as its elements arrive. On a database query, `FirstOrNone`, `FirstOrError`, `SingleOrNone` and `ElementAtOrNone` ask the provider for only the element they need, and let it run the predicate (a `WHERE` and a `LIMIT` in @@ -225,6 +226,20 @@ sequence order. `Traverse` calls nothing after it. None of them collects errors. cancellation token you give it is cancelled. For operations that may all run at once, use `Task.WhenAll` and `Sequence()`. +An async stream (`IAsyncEnumerable`), such as a paged API or a query read with +`AsAsyncEnumerable()`, needs no list first: + +```csharp +IAsyncEnumerable known = lines.Choose(line => Sku.TryFrom(line)); // the lines that are SKUs, as they arrive +Result, SkuFault> imported = await rows.SequenceAsync(cancellationToken); // every SKU, or the first fault +Result, OrderFault> found = await ids.TraverseAsync(orders.FindAsync); // one lookup at a time, in order +``` + +`SequenceAsync` and `TraverseAsync` stop at the first error and read nothing after it. All three read +the stream in order, and dispose its enumerator however they stop: at an error, at a `break`, or when +your selector throws. The token you give `SequenceAsync` or `TraverseAsync` reaches the stream. For +`Choose`, pass yours to `WithCancellation` on what it returns. + The asynchronous steps take callbacks that return a `Task`, and wait for them. `.TapErrorAsync(fault => audit.RecordAsync(fault))` has recorded the fault when the chain's `await` returns, and what it throws comes out of that `await`. diff --git a/src/CodoMetis.TypeKit/ResultExtensions.cs b/src/CodoMetis.TypeKit/ResultExtensions.cs index a4394bea..eb0a0978 100644 --- a/src/CodoMetis.TypeKit/ResultExtensions.cs +++ b/src/CodoMetis.TypeKit/ResultExtensions.cs @@ -1385,6 +1385,138 @@ public async Task> TraverseAsync(FuncThe stream. + /// The type of the elements. + extension(IAsyncEnumerable source) + { + /// + /// Applies an asynchronous operation that may fail to every element of a stream, one at a time + /// and in order, stopping at the first error: the next element is not read, nor its operation + /// started, after one failed, and the stream's enumerator is disposed. + /// + /// Called with each element and awaited, until one fails. + /// The type of each produced value. + /// The type of the error. + /// A success with every produced value, in order, or the first error. + /// returned an uninitialized result. + /// returned no task. + public async Task, TError>> TraverseAsync(Func>> selector) + where TResult : notnull + where TError : notnull + { + ArgumentNullException.ThrowIfNull(source); + ArgumentNullException.ThrowIfNull(selector); + + var values = new List(); + + await foreach (var element in source.ConfigureAwait(false)) + { + var result = await ThrowHelper.Awaitable(selector(element), nameof(TraverseAsync), nameof(selector)).ConfigureAwait(false); + if (!result.TryGetValue(out var value, out var error)) return Result, TError>.Error(error); + + values.Add(value); + } + + return Result, TError>.Success(values); + } + + /// + /// Applies an asynchronous operation that may fail to every element of a stream, one at a time + /// and in order, stopping at the first error, with a cancellation token handed to the stream and + /// to each operation, and checked before each element is read and before each operation: once + /// cancelled, nothing more is read or started and the + /// propagates, since a cancellation is not an error. + /// + /// Called with each element and the token, and awaited, until one fails. + /// Handed to the stream's enumerator and to each operation, and checked before each. + /// The type of each produced value. + /// The type of the error. + /// A success with every produced value, in order, or the first error. + /// returned an uninitialized result. + /// returned no task. + /// was cancelled before an element was read or an operation started. + public async Task, TError>> TraverseAsync(Func>> selector, CancellationToken cancellationToken) + where TResult : notnull + where TError : notnull + { + ArgumentNullException.ThrowIfNull(source); + ArgumentNullException.ThrowIfNull(selector); + + var values = new List(); + + cancellationToken.ThrowIfCancellationRequested(); + await foreach (var element in source.WithCancellation(cancellationToken).ConfigureAwait(false)) + { + cancellationToken.ThrowIfCancellationRequested(); + + var result = await ThrowHelper.Awaitable(selector(element, cancellationToken), nameof(TraverseAsync), nameof(selector)).ConfigureAwait(false); + if (!result.TryGetValue(out var value, out var error)) return Result, TError>.Error(error); + + values.Add(value); + cancellationToken.ThrowIfCancellationRequested(); + } + + return Result, TError>.Success(values); + } + + /// + /// Applies an asynchronous command that may fail to every element of a stream, one at a time and + /// in order, stopping at the first error: the next element is not read, nor its command started, + /// after one failed, and the stream's enumerator is disposed. + /// + /// Called with each element and awaited, until one fails. + /// The type of the error. + /// A success if every command succeeded, or the first error. + /// returned an uninitialized result. + /// returned no task. + public async Task> TraverseAsync(Func>> selector) where TError : notnull + { + ArgumentNullException.ThrowIfNull(source); + ArgumentNullException.ThrowIfNull(selector); + + await foreach (var element in source.ConfigureAwait(false)) + { + var result = await ThrowHelper.Awaitable(selector(element), nameof(TraverseAsync), nameof(selector)).ConfigureAwait(false); + if (result.TryGetError(out var error)) return Result.Error(error); + } + + return Result.Success(); + } + + /// + /// Applies an asynchronous command that may fail to every element of a stream, one at a time and + /// in order, stopping at the first error, with a cancellation token handed to the stream and to + /// each command, and checked before each element is read and before each command: once + /// cancelled, nothing more is read or started and the + /// propagates, since a cancellation is not an error. + /// + /// Called with each element and the token, and awaited, until one fails. + /// Handed to the stream's enumerator and to each command, and checked before each. + /// The type of the error. + /// A success if every command succeeded, or the first error. + /// returned an uninitialized result. + /// returned no task. + /// was cancelled before an element was read or a command started. + public async Task> TraverseAsync(Func>> selector, CancellationToken cancellationToken) where TError : notnull + { + ArgumentNullException.ThrowIfNull(source); + ArgumentNullException.ThrowIfNull(selector); + + cancellationToken.ThrowIfCancellationRequested(); + await foreach (var element in source.WithCancellation(cancellationToken).ConfigureAwait(false)) + { + cancellationToken.ThrowIfCancellationRequested(); + + var result = await ThrowHelper.Awaitable(selector(element, cancellationToken), nameof(TraverseAsync), nameof(selector)).ConfigureAwait(false); + if (result.TryGetError(out var error)) return Result.Error(error); + + cancellationToken.ThrowIfCancellationRequested(); + } + + return Result.Success(); + } + } + /// The results. /// The type of the values. /// The type of the error. @@ -1435,6 +1567,66 @@ public Result Sequence() } } + /// The stream of results. + /// The type of the values. + /// The type of the error. + extension(IAsyncEnumerable> results) where T : notnull where TError : notnull + { + /// Every value of a stream, if every result in it is a success. + /// Handed to the stream's enumerator, and checked before each result is read. + /// + /// A success with every value, in order, or the first error. After an error nothing more is read + /// and the stream's enumerator is disposed. + /// + /// A result up to the first error is uninitialized. + /// was cancelled before a result was read. + public async Task, TError>> SequenceAsync(CancellationToken cancellationToken = default) + { + ArgumentNullException.ThrowIfNull(results); + + var values = new List(); + + cancellationToken.ThrowIfCancellationRequested(); + await foreach (var result in results.WithCancellation(cancellationToken).ConfigureAwait(false)) + { + if (!result.TryGetValue(out var value, out var error)) return Result, TError>.Error(error); + + values.Add(value); + cancellationToken.ThrowIfCancellationRequested(); + } + + return Result, TError>.Success(values); + } + } + + /// The stream of results. + /// The type of the error. + extension(IAsyncEnumerable> results) where TError : notnull + { + /// A success if every result of a stream is one. + /// Handed to the stream's enumerator, and checked before each result is read. + /// + /// A success, or the first error. After an error nothing more is read and the stream's enumerator + /// is disposed. + /// + /// A result up to the first error is uninitialized. + /// was cancelled before a result was read. + public async Task> SequenceAsync(CancellationToken cancellationToken = default) + { + ArgumentNullException.ThrowIfNull(results); + + cancellationToken.ThrowIfCancellationRequested(); + await foreach (var result in results.WithCancellation(cancellationToken).ConfigureAwait(false)) + { + if (result.TryGetError(out var error)) return Result.Error(error); + + cancellationToken.ThrowIfCancellationRequested(); + } + + return Result.Success(); + } + } + /// An optional input that is validated when present. /// The type of the value. /// The type of the error. diff --git a/src/CodoMetis.TypeKit/SequenceOptionExtensions.cs b/src/CodoMetis.TypeKit/SequenceOptionExtensions.cs index d74e9935..7f10c7dd 100644 --- a/src/CodoMetis.TypeKit/SequenceOptionExtensions.cs +++ b/src/CodoMetis.TypeKit/SequenceOptionExtensions.cs @@ -1,7 +1,7 @@ namespace CodoMetis.TypeKit; /// -/// Choose(selector) over a sequence whose elements may be null: only what the selector finds +/// Choose(selector) over a sequence or a stream whose elements may be null: only what the selector finds /// must be non-null, so a sequence of string? or int? is chosen from without a warning, /// its nulls the selector's to answer. The other sequence members, Choose() over options among /// them, are on . @@ -35,4 +35,26 @@ public IEnumerable Choose(Func> selector) w return Option.Chosen(source, selector); } } + + /// The stream, whose elements may be null. + /// The type of the elements, nullable or not. + extension(IAsyncEnumerable source) + { + /// + /// The values finds in a stream, as they arrive: for each element it + /// answers with Some, that value, in order, and nothing for a None. Lazy, as an async + /// LINQ operator is, and the token given to WithCancellation or GetAsyncEnumerator + /// reaches the stream it reads. + /// + /// Called with each element as the stream is read. + /// The type of the values. + /// The values found, in order. + public IAsyncEnumerable Choose(Func> selector) where TResult : notnull + { + ArgumentNullException.ThrowIfNull(source); + ArgumentNullException.ThrowIfNull(selector); + + return Option.Chosen(source, selector); + } + } } diff --git a/test/CodoMetis.TypeKit.Analyzers.Tests/IgnoredOutcomeAnalyzerTests.cs b/test/CodoMetis.TypeKit.Analyzers.Tests/IgnoredOutcomeAnalyzerTests.cs index a5d51f73..0dbbd27f 100644 --- a/test/CodoMetis.TypeKit.Analyzers.Tests/IgnoredOutcomeAnalyzerTests.cs +++ b/test/CodoMetis.TypeKit.Analyzers.Tests/IgnoredOutcomeAnalyzerTests.cs @@ -651,6 +651,50 @@ public static Task Save(Service service) Cmtk0003(3, "Result", "LoadAsync")) .RunAsync(TestContext.Current.CancellationToken); + /// + /// SequenceAsync and TraverseAsync over an async stream answer a Task of a result, + /// so dropping one reports as dropping TraverseAsync over a sequence does: awaited or not, and + /// converted to a plain Task. Kept or discarded with _ =, it is silent. + /// + [Fact] + public Task A_dropped_SequenceAsync_or_TraverseAsync_over_a_stream_reports_as_over_a_sequence() => + Test( + """ + public static class Caller + { + public static async Task Run(Service service, int[] ids, IAsyncEnumerable stream, IAsyncEnumerable> values) + { + {|#0:await ids.TraverseAsync(service.CancelAsync)|}; + {|#1:await stream.TraverseAsync(service.CancelAsync)|}; + {|#2:await stream.TraverseAsync((id, token) => service.LoadAsync(), default)|}; + {|#3:await service.SaveEachAsync(ids).SequenceAsync()|}; + {|#4:await values.SequenceAsync(default)|}; + #pragma warning disable CS4014 + {|#5:service.SaveEachAsync(ids).SequenceAsync()|}; + {|#6:stream.TraverseAsync(service.CancelAsync)|}; + #pragma warning restore CS4014 + + _ = await values.SequenceAsync(); + var all = await stream.TraverseAsync(id => service.LoadAsync()); + if (await service.SaveEachAsync(ids).SequenceAsync()) { } + } + + public static Task Drop(Service service, int[] ids) => {|#7:service.SaveEachAsync(ids).SequenceAsync()|}; + + public static Task DropEach(Service service, IAsyncEnumerable stream) => {|#8:stream.TraverseAsync(service.CancelAsync)|}; + } + """, + Cmtk0003(0, "Result", "TraverseAsync"), + Cmtk0003(1, "Result", "TraverseAsync"), + Cmtk0003(2, "Result", "TraverseAsync"), + Cmtk0003(3, "Result", "SequenceAsync"), + Cmtk0003(4, "Result", "SequenceAsync"), + Cmtk0003(5, "Result", "SequenceAsync"), + Cmtk0003(6, "Result", "TraverseAsync"), + Cmtk0003(7, "Result", "SequenceAsync"), + Cmtk0003(8, "Result", "TraverseAsync")) + .RunAsync(TestContext.Current.CancellationToken); + /// A task that keeps its result type, an explicit cast, and a task the caller holds are not dropped by the conversion. [Fact] public Task A_task_of_a_result_that_keeps_its_type_is_cast_or_is_held_stays_silent() => diff --git a/test/CodoMetis.TypeKit.Samples/Base/ResultSamples.cs b/test/CodoMetis.TypeKit.Samples/Base/ResultSamples.cs index 219d70db..d1481893 100644 --- a/test/CodoMetis.TypeKit.Samples/Base/ResultSamples.cs +++ b/test/CodoMetis.TypeKit.Samples/Base/ResultSamples.cs @@ -83,6 +83,21 @@ from price in pricing.For(item, customer) Result, SkuFault> all = parsed.Sequence(); // end sample } + + public static async Task<(List, Result, SkuFault>, Result, OrderFault>)> Streams( + IOrders orders, + IAsyncEnumerable lines, + IAsyncEnumerable> rows, + IAsyncEnumerable ids, + CancellationToken cancellationToken) + { + // sample: CodoMetis.TypeKit/streams + IAsyncEnumerable known = lines.Choose(line => Sku.TryFrom(line)); // the lines that are SKUs, as they arrive + Result, SkuFault> imported = await rows.SequenceAsync(cancellationToken); // every SKU, or the first fault + Result, OrderFault> found = await ids.TraverseAsync(orders.FindAsync); // one lookup at a time, in order + // end sample + return (await known.ToListAsync(cancellationToken), imported, found); + } } public readonly partial record struct OrderId : IValue; diff --git a/test/CodoMetis.TypeKit.Tests/AsyncStreamTests.cs b/test/CodoMetis.TypeKit.Tests/AsyncStreamTests.cs new file mode 100644 index 00000000..148e4f8d --- /dev/null +++ b/test/CodoMetis.TypeKit.Tests/AsyncStreamTests.cs @@ -0,0 +1,445 @@ +namespace CodoMetis.TypeKit.Tests; + +/// +/// Choose, SequenceAsync and TraverseAsync over an : +/// in order, the caller's token reaching the stream's enumerator, nothing read after the first error, +/// and the enumerator disposed however the reading ends. +/// +/// +/// Every case runs over a stream whose elements arrive at once and over one whose elements each wait, +/// since an async iterator that never waits completes every step synchronously and hides a step that +/// resumes in the wrong place. +/// +public sealed class AsyncStreamTests +{ + public static TheoryData Waits => [false, true]; + + private static Option Even(int n) => n % 2 == 0 ? Option.Some(n) : Option.None(); + + private static Result Ok(int value) => Result.Success(value); + + private static Result Fail(string error) => Result.Error(error); + + [Theory] + [MemberData(nameof(Waits))] + public async Task Choose_keeps_the_values_of_the_options_that_hold_one_in_order(bool waits) + { + var options = new TrackedStream>(waits, Option.Some(1), Option.None(), Option.Some(3), default, Option.Some(5)); + + (await options.Choose().ToListAsync(TestContext.Current.CancellationToken)).ShouldBe([1, 3, 5]); + (await new TrackedStream>(waits).Choose().ToListAsync(TestContext.Current.CancellationToken)).ShouldBeEmpty(); + options.Disposed.ShouldBeTrue(); + } + + /// A default option is None, so a stream holding one answers what the same sequence does. + [Theory] + [MemberData(nameof(Waits))] + public async Task Choose_over_a_stream_answers_what_it_answers_over_the_same_sequence(bool waits) + { + Option[] options = [default, Option.Some(2), Option.None(), default, Option.Some(4)]; + int[] numbers = [1, 2, 3, 4, 5, 6]; + + (await new TrackedStream>(waits, options).Choose().ToListAsync(TestContext.Current.CancellationToken)).ShouldBe(options.Choose().ToList()); + (await new TrackedStream(waits, numbers).Choose(Even).ToListAsync(TestContext.Current.CancellationToken)).ShouldBe(numbers.Choose(Even).ToList()); + } + + [Theory] + [MemberData(nameof(Waits))] + public async Task Choose_with_a_selector_keeps_what_it_answers_Some_for_and_reads_null_elements(bool waits) + { + (await new TrackedStream(waits, 1, 2, 3, 4, 5, 6).Choose(Even).ToListAsync(TestContext.Current.CancellationToken)).ShouldBe([2, 4, 6]); + (await new TrackedStream(waits, "Annie", null, "Bo").Choose(nickname => nickname.ToOption()).ToListAsync(TestContext.Current.CancellationToken)).ShouldBe(["Annie", "Bo"]); + (await new TrackedStream(waits, 3, null, 5).Choose(floor => floor.ToOption()).ToListAsync(TestContext.Current.CancellationToken)).ShouldBe([3, 5]); + } + + /// An async LINQ operator reads its source when it is enumerated, not when it is called, and only as far as the caller reads. + [Theory] + [MemberData(nameof(Waits))] + public async Task Choose_is_lazy_and_disposes_the_stream_when_the_caller_stops(bool waits) + { + var numbers = new TrackedStream(waits, 1, 2, 3, 4, 5, 6); + + var chosen = numbers.Choose(Even); + numbers.Enumerations.ShouldBe(0, "nothing is read until the result is enumerated"); + + await foreach (var value in chosen) + { + value.ShouldBe(2); + break; + } + + numbers.Read.ShouldBe(2, "the stream is read only as far as the first value found"); + numbers.Disposed.ShouldBeTrue("a break disposes the stream's enumerator"); + } + + /// The arguments are checked when the call is made, as LINQ's are, not when the stream is first read. + [Fact] + public void Choose_checks_its_arguments_before_the_stream_is_read() + { + Should.Throw(() => ((IAsyncEnumerable)null!).Choose(Even)).ParamName.ShouldBe("source"); + Should.Throw(() => ((IAsyncEnumerable>)null!).Choose()).ParamName.ShouldBe("options"); + Should.Throw(() => new TrackedStream(false, 1).Choose((Func>)null!)).ParamName.ShouldBe("selector"); + } + + [Theory] + [MemberData(nameof(Waits))] + public async Task Choose_hands_the_token_of_WithCancellation_and_of_GetAsyncEnumerator_to_the_stream(bool waits) + { + using var cancellation = new CancellationTokenSource(); + var numbers = new TrackedStream(waits, 1, 2); + var options = new TrackedStream>(waits, Option.Some(1)); + + await foreach (var _ in numbers.Choose(Even).WithCancellation(cancellation.Token)) { } + numbers.Token.ShouldBe(cancellation.Token); + + await foreach (var _ in options.Choose().WithCancellation(cancellation.Token)) { } + options.Token.ShouldBe(cancellation.Token); + + var held = new TrackedStream(waits, 1, 2, 3, 4); + await using (var enumerator = held.Choose(Even).GetAsyncEnumerator(cancellation.Token)) + { + (await enumerator.MoveNextAsync()).ShouldBeTrue(); + held.Disposed.ShouldBeFalse(); + } + + held.Token.ShouldBe(cancellation.Token); + held.Disposed.ShouldBeTrue("disposing the chosen values disposes the stream's enumerator"); + } + + /// A stream that honours its token stops the chosen values when the caller cancels. + [Theory] + [MemberData(nameof(Waits))] + public async Task Choose_stops_when_the_caller_cancels_a_stream_that_honours_its_token(bool waits) + { + using var cancellation = new CancellationTokenSource(); + var numbers = new TrackedStream(waits, 2, 4, 6) { OnRead = n => { if (n == 4) cancellation.Cancel(); }, HonoursToken = true }; + var seen = new List(); + + await Should.ThrowAsync(async () => + { + await foreach (var value in numbers.Choose(Even).WithCancellation(cancellation.Token)) seen.Add(value); + }); + + seen.ShouldBe([2, 4]); + numbers.Disposed.ShouldBeTrue(); + } + + [Theory] + [MemberData(nameof(Waits))] + public async Task Choose_disposes_the_stream_when_the_selector_throws(bool waits) + { + var numbers = new TrackedStream(waits, 1, 2, 3); + + await Should.ThrowAsync(() => numbers.Choose(n => n == 2 ? throw new FormatException() : Option.Some(n)).ToListAsync(TestContext.Current.CancellationToken).AsTask()); + + numbers.Read.ShouldBe(2); + numbers.Disposed.ShouldBeTrue(); + } + + [Theory] + [MemberData(nameof(Waits))] + public async Task SequenceAsync_collects_every_value_in_order(bool waits) + { + var values = await new TrackedStream>(waits, Ok(1), Ok(2), Ok(3)).SequenceAsync(TestContext.Current.CancellationToken); + values.Match(list => list, _ => []).ShouldBe([1, 2, 3]); + + var none = await new TrackedStream>(waits).SequenceAsync(TestContext.Current.CancellationToken); + none.Match(list => list.Count, _ => -1).ShouldBe(0); + + ((bool)await new TrackedStream>(waits, Result.Success(), Result.Success()).SequenceAsync(TestContext.Current.CancellationToken)).ShouldBeTrue(); + ((bool)await new TrackedStream>(waits).SequenceAsync(TestContext.Current.CancellationToken)).ShouldBeTrue(); + } + + [Theory] + [MemberData(nameof(Waits))] + public async Task SequenceAsync_answers_the_first_error_reads_nothing_after_it_and_disposes_the_stream(bool waits) + { + var results = new TrackedStream>(waits, Ok(1), Fail("x"), Fail("y"), Ok(4)); + + (await results.SequenceAsync(TestContext.Current.CancellationToken)).Match(_ => "", error => error).ShouldBe("x"); + results.Read.ShouldBe(2); + results.Disposed.ShouldBeTrue(); + + var commands = new TrackedStream>(waits, Result.Success(), Result.Error("refused"), default); + + (await commands.SequenceAsync(TestContext.Current.CancellationToken)).TryGetError(out var refused).ShouldBeTrue(); + refused.ShouldBe("refused"); + commands.Read.ShouldBe(2, "the uninitialized result after the error is never read"); + commands.Disposed.ShouldBeTrue(); + } + + [Theory] + [MemberData(nameof(Waits))] + public async Task SequenceAsync_refuses_an_uninitialized_result_and_disposes_the_stream(bool waits) + { + var results = new TrackedStream>(waits, Ok(1), default, Ok(3)); + var commands = new TrackedStream>(waits, Result.Success(), default); + + (await Should.ThrowAsync(() => results.SequenceAsync())).Message.ShouldContain("never initialized"); + (await Should.ThrowAsync(() => commands.SequenceAsync())).Message.ShouldContain("never initialized"); + results.Disposed.ShouldBeTrue(); + commands.Disposed.ShouldBeTrue(); + } + + /// + /// The token reaches the stream's enumerator and is checked before each result is read, so a stream + /// that ignores its token still stops: the cancellation propagates rather than becoming an error. + /// + [Theory] + [MemberData(nameof(Waits))] + public async Task SequenceAsync_hands_the_token_to_the_stream_and_stops_once_cancelled(bool waits) + { + using var cancellation = new CancellationTokenSource(); + var results = new TrackedStream>(waits, Ok(1), Ok(2), Ok(3)) { OnRead = r => { if (r == Ok(2)) cancellation.Cancel(); } }; + + await Should.ThrowAsync(() => results.SequenceAsync(cancellation.Token)); + results.Token.ShouldBe(cancellation.Token); + results.Read.ShouldBe(2, "nothing is read after the token was cancelled"); + results.Disposed.ShouldBeTrue(); + + using var commandCancellation = new CancellationTokenSource(); + var commands = new TrackedStream>(waits, Result.Success(), Result.Success(), Result.Success()); + commands.OnRead = _ => { if (commands.Read == 2) commandCancellation.Cancel(); }; + + await Should.ThrowAsync(() => commands.SequenceAsync(commandCancellation.Token)); + commands.Token.ShouldBe(commandCancellation.Token); + commands.Read.ShouldBe(2); + commands.Disposed.ShouldBeTrue(); + } + + [Fact] + public async Task SequenceAsync_with_a_cancelled_token_reads_nothing() + { + var results = new TrackedStream>(false, Ok(1)); + var commands = new TrackedStream>(false, Result.Success()); + + await Should.ThrowAsync(() => results.SequenceAsync(new CancellationToken(canceled: true))); + await Should.ThrowAsync(() => commands.SequenceAsync(new CancellationToken(canceled: true))); + results.Read.ShouldBe(0); + commands.Read.ShouldBe(0); + } + + [Theory] + [MemberData(nameof(Waits))] + public async Task TraverseAsync_awaits_each_operation_in_order_and_collects_the_values(bool waits) + { + var order = new List(); + var texts = new TrackedStream(waits, "1", "2", "3") { OnRead = text => order.Add($"read {text}") }; + + async Task> Parse(string text) + { + order.Add($"start {text}"); + await Task.Yield(); + order.Add($"end {text}"); + return int.TryParse(text, out var n) ? Ok(n) : Fail(text); + } + + var parsed = await texts.TraverseAsync(Parse); + + parsed.Match(values => values, _ => []).ShouldBe([1, 2, 3]); + order.ShouldBe(["read 1", "start 1", "end 1", "read 2", "start 2", "end 2", "read 3", "start 3", "end 3"], "one element at a time, its operation finished before the next is read"); + texts.Disposed.ShouldBeTrue(); + } + + [Theory] + [MemberData(nameof(Waits))] + public async Task TraverseAsync_stops_reading_after_an_error_and_disposes_the_stream(bool waits) + { + var started = new List(); + var texts = new TrackedStream(waits, "1", "x", "y", "4"); + + Task> Parse(string text) + { + started.Add(text); + return Task.FromResult(int.TryParse(text, out var n) ? Ok(n) : Fail(text)); + } + + (await texts.TraverseAsync(Parse)).Match(_ => "", error => error).ShouldBe("x"); + started.ShouldBe(["1", "x"]); + texts.Read.ShouldBe(2); + texts.Disposed.ShouldBeTrue(); + + var sent = new List(); + var numbers = new TrackedStream(waits, 1, 2, 3, 4); + + Task> Send(int n) + { + if (n == 3) return Task.FromResult(Result.Error($"refused {n}")); + + sent.Add(n); + return Task.FromResult(Result.Success()); + } + + (await numbers.TraverseAsync(Send)).TryGetError(out var error).ShouldBeTrue(); + error.ShouldBe("refused 3"); + sent.ShouldBe([1, 2]); + numbers.Read.ShouldBe(3); + numbers.Disposed.ShouldBeTrue(); + + ((bool)await new TrackedStream(waits, 1, 2).TraverseAsync(Send)).ShouldBeTrue(); + ((bool)await new TrackedStream(waits).TraverseAsync(Send)).ShouldBeTrue(); + (await new TrackedStream(waits).TraverseAsync(Parse)).Match(list => list.Count, _ => -1).ShouldBe(0); + } + + [Theory] + [MemberData(nameof(Waits))] + public async Task TraverseAsync_disposes_the_stream_when_the_selector_throws(bool waits) + { + var numbers = new TrackedStream(waits, 1, 2, 3); + var commands = new TrackedStream(waits, 1, 2, 3); + + await Should.ThrowAsync(() => numbers.TraverseAsync(n => n == 2 ? throw new FormatException() : Task.FromResult(Ok(n)))); + await Should.ThrowAsync(() => commands.TraverseAsync((n, _) => n == 2 ? throw new FormatException() : Task.FromResult(Result.Success()), CancellationToken.None)); + + numbers.Read.ShouldBe(2); + numbers.Disposed.ShouldBeTrue(); + commands.Read.ShouldBe(2); + commands.Disposed.ShouldBeTrue(); + } + + /// + /// The token is handed to the stream and to each operation, and checked before each element is read + /// and before each operation: once cancelled, nothing more is read or started, and the cancellation + /// propagates rather than becoming an error. + /// + [Theory] + [MemberData(nameof(Waits))] + public async Task TraverseAsync_hands_the_token_on_and_stops_once_cancelled(bool waits) + { + using var cancellation = new CancellationTokenSource(); + var numbers = new TrackedStream(waits, 1, 2, 3); + var started = new List(); + + Task> Load(int n, CancellationToken token) + { + token.ShouldBe(cancellation.Token, "the operation receives the caller's token"); + started.Add(n); + if (n == 2) cancellation.Cancel(); + return Task.FromResult(Ok(n)); + } + + await Should.ThrowAsync(() => numbers.TraverseAsync(Load, cancellation.Token)); + numbers.Token.ShouldBe(cancellation.Token, "the stream's enumerator receives the caller's token"); + started.ShouldBe([1, 2]); + numbers.Read.ShouldBe(2, "the third element is not read after the token was cancelled"); + numbers.Disposed.ShouldBeTrue(); + + using var commandCancellation = new CancellationTokenSource(); + var commands = new TrackedStream(waits, 1, 2, 3) { OnRead = n => { if (n == 2) commandCancellation.Cancel(); } }; + started.Clear(); + + Task> Send(int n, CancellationToken token) + { + token.ShouldBe(commandCancellation.Token); + started.Add(n); + return Task.FromResult(Result.Success()); + } + + await Should.ThrowAsync(() => commands.TraverseAsync(Send, commandCancellation.Token)); + commands.Token.ShouldBe(commandCancellation.Token); + started.ShouldBe([1], "the command for an element read after the token was cancelled is not started"); + commands.Disposed.ShouldBeTrue(); + } + + [Fact] + public async Task TraverseAsync_with_a_cancelled_token_reads_and_starts_nothing() + { + var numbers = new TrackedStream(false, 1); + var started = 0; + + await Should.ThrowAsync(() => numbers.TraverseAsync((n, _) => { started++; return Task.FromResult(Ok(n)); }, new CancellationToken(canceled: true))); + await Should.ThrowAsync(() => numbers.TraverseAsync((_, _) => { started++; return Task.FromResult(Result.Success()); }, new CancellationToken(canceled: true))); + + started.ShouldBe(0); + numbers.Read.ShouldBe(0); + } + + [Theory] + [MemberData(nameof(Waits))] + public async Task TraverseAsync_with_a_token_that_is_not_cancelled_runs_every_operation(bool waits) + { + var values = await new TrackedStream(waits, 1, 2).TraverseAsync((n, _) => Task.FromResult(Ok(n * 10)), CancellationToken.None); + values.Match(list => list, _ => []).ShouldBe([10, 20]); + + var command = await new TrackedStream(waits, 1, 2).TraverseAsync((_, _) => Task.FromResult(Result.Success()), CancellationToken.None); + ((bool)command).ShouldBeTrue(); + + var refused = new TrackedStream(waits, 1, 2, 3); + (await refused.TraverseAsync((n, _) => Task.FromResult(n == 2 ? Fail("two") : Ok(n)), CancellationToken.None)).Match(_ => "", error => error).ShouldBe("two"); + refused.Read.ShouldBe(2); + refused.Disposed.ShouldBeTrue(); + } + + /// The calls in a consumer's compilation, nullable annotations on, beside System.Linq.AsyncEnumerable's operators: no warning and no ambiguity. + [Theory] + [InlineData("IAsyncEnumerable nicknames", "nicknames.Choose(nickname => nickname.ToOption())")] + [InlineData("IAsyncEnumerable floors", "floors.Where(floor => floor > 0).Choose(floor => floor.ToOption()).Select(floor => floor * 10)")] + [InlineData("IAsyncEnumerable numbers", "numbers.Select(n => n > 0 ? Option.Some(n) : Option.None()).Choose()")] + [InlineData("IAsyncEnumerable> results", "results.SequenceAsync()")] + [InlineData("IAsyncEnumerable> results", "results.Where(result => true).SequenceAsync(CancellationToken.None)")] + [InlineData("IAsyncEnumerable texts", "texts.TraverseAsync(text => Task.FromResult(Result.Success(text.Length)))")] + [InlineData("IAsyncEnumerable texts", "texts.TraverseAsync(async (text, token) => { await Task.Delay(1, token); return Result.Success(); }, CancellationToken.None)")] + [InlineData("int[] numbers", "numbers.Choose(n => Option.Some(n))")] + [InlineData("int[] numbers", "numbers.TraverseAsync(n => Task.FromResult(Result.Success(n)))")] + public void The_stream_members_compile_without_a_warning_beside_the_async_LINQ_operators(string parameter, string call) => + ConsumerCompilation.Of($$""" + using System.Collections.Generic; + using System.Linq; + using System.Threading; + using System.Threading.Tasks; + using CodoMetis.TypeKit; + + public static class Checks + { + public static object Run({{parameter}}) => {{call}}; + } + """).Problems().ShouldBeEmpty(); + + /// + /// A stream that records what its enumerator was given and how far it was read, and whether it was + /// disposed. It ignores its token unless told to honour it, as many producers do. + /// + private sealed class TrackedStream(bool waits, params T[] elements) : IAsyncEnumerable + { + public CancellationToken Token { get; private set; } + + public int Enumerations { get; private set; } + + /// How many elements were handed out. + public int Read { get; private set; } + + public bool Disposed { get; private set; } + + public Action? OnRead { get; set; } + + public bool HonoursToken { get; init; } + + public IAsyncEnumerator GetAsyncEnumerator(CancellationToken cancellationToken = default) + { + Token = cancellationToken; + Enumerations++; + + return Enumerate(cancellationToken); + } + + private async IAsyncEnumerator Enumerate(CancellationToken cancellationToken) + { + try + { + foreach (var element in elements) + { + if (waits) await Task.Yield(); + if (HonoursToken) cancellationToken.ThrowIfCancellationRequested(); + + Read++; + OnRead?.Invoke(element); + yield return element; + } + } + finally + { + Disposed = true; + } + } + } +} diff --git a/test/CodoMetis.TypeKit.Tests/AsynchronousContextTests.cs b/test/CodoMetis.TypeKit.Tests/AsynchronousContextTests.cs index f412579f..7a9f7ce8 100644 --- a/test/CodoMetis.TypeKit.Tests/AsynchronousContextTests.cs +++ b/test/CodoMetis.TypeKit.Tests/AsynchronousContextTests.cs @@ -20,7 +20,7 @@ namespace CodoMetis.TypeKit.Tests; /// (.Result, .Wait(), GetAwaiter().GetResult()) never does. A member that resumed on it /// would hang that caller, which is why ConfigureAwait(false) stays; such a post fails here at once. /// The cases are written by hand, and the completeness test holds them to every public member of the -/// assembly that returns a task. +/// assembly that returns a task or an , whose cases read it to the end. /// /// public sealed class AsynchronousContextTests @@ -72,6 +72,21 @@ public sealed class AsynchronousContextTests ["Result.TraverseAsync(IEnumerable, Func>>, CancellationToken)"] = p => new[] { 1, 2 }.TraverseAsync((x, _) => x == 1 ? p.First(Ok) : p.Later(p.Seen(Ok)), CancellationToken.None), + // A stream's first element is what the rest wait for, so the member's own await of the stream is + // the one that would resume on the caller's context; a selector runs as each element arrives. + ["Result.TraverseAsync(IAsyncEnumerable, Func>>)"] = + p => p.Stream(1, 2).TraverseAsync(x => x == 1 ? p.First(Value) : p.Later(p.Seen(Value))), + ["Result.TraverseAsync(IAsyncEnumerable, Func>>, CancellationToken)"] = + p => p.Stream(1, 2).TraverseAsync((x, _) => x == 1 ? p.First(Value) : p.Later(p.Seen(Value)), CancellationToken.None), + ["Result.TraverseAsync(IAsyncEnumerable, Func>>)"] = + p => p.Stream(1, 2).TraverseAsync(x => x == 1 ? p.First(Ok) : p.Later(p.Seen(Ok))), + ["Result.TraverseAsync(IAsyncEnumerable, Func>>, CancellationToken)"] = + p => p.Stream(1, 2).TraverseAsync((x, _) => x == 1 ? p.First(Ok) : p.Later(p.Seen(Ok)), CancellationToken.None), + ["Result.SequenceAsync(IAsyncEnumerable>, CancellationToken)"] = p => p.Stream(Value, Value).SequenceAsync(), + ["Result.SequenceAsync(IAsyncEnumerable>, CancellationToken)"] = p => p.Stream(Ok, Ok).SequenceAsync(), + ["Option.Choose(IAsyncEnumerable>)"] = p => Drain(p.Stream(Some, Some).Choose()), + ["SequenceOptionExtensions.Choose(IAsyncEnumerable, Func>)"] = p => Drain(p.Stream(1, 2).Choose(x => p.Seen(Some))), + // The delegate runs at once, on the caller's context; the task it returns is awaited without it. ["Result.CatchAsync(Func>)"] = p => Result.CatchAsync(() => p.Later(p.Seen(1))), ["Result.CatchAsync(Func)"] = p => Result.CatchAsync(() => (Task)p.Later(p.Seen(1))), @@ -117,7 +132,13 @@ public sealed class AsynchronousContextTests }; /// Members that take no callback: only that nothing resumes on the caller's context is checked. - private static readonly HashSet WithoutCallback = ["Option.ToResultAsync(Task>, TError)"]; + private static readonly HashSet WithoutCallback = + [ + "Option.ToResultAsync(Task>, TError)", + "Result.SequenceAsync(IAsyncEnumerable>, CancellationToken)", + "Result.SequenceAsync(IAsyncEnumerable>, CancellationToken)", + "Option.Choose(IAsyncEnumerable>)", + ]; public static TheoryData CaseNames => [.. Cases.Keys]; @@ -173,15 +194,21 @@ await caller.Run(async () => } [Fact] - public void Every_public_member_that_returns_a_task_has_a_case() + public void Every_public_member_that_returns_a_task_or_a_stream_has_a_case() { var members = AsynchronousMembers().ToList(); members.Count.ShouldBeGreaterThanOrEqualTo(50); - members.Except(Cases.Keys).ShouldBeEmpty("These public members return a task and have no case here."); + members.Except(Cases.Keys).ShouldBeEmpty("These public members return a task or a stream and have no case here."); Cases.Keys.Except(members).ShouldBeEmpty("These cases name no public member of the assembly."); } + /// Reads a stream to the end without capturing the caller's context, as a member's own awaits must not. + private static async Task Drain(IAsyncEnumerable stream) + { + await foreach (var _ in stream.ConfigureAwait(false)) { } + } + private static Task Start(Func chain, Probe probe, SynchronizationContext caller) { var previous = SynchronizationContext.Current; @@ -202,7 +229,7 @@ private static IEnumerable AsynchronousMembers() from type in typeof(Option<>).Assembly.GetExportedTypes() where !type.Name.Contains('<') && type.GetCustomAttribute() is null from method in type.GetMethods(BindingFlags.Public | BindingFlags.Instance | BindingFlags.Static | BindingFlags.DeclaredOnly) - where typeof(Task).IsAssignableFrom(method.ReturnType) + where typeof(Task).IsAssignableFrom(method.ReturnType) || method.ReturnType.IsGenericType && method.ReturnType.GetGenericTypeDefinition() == typeof(IAsyncEnumerable<>) where method.GetCustomAttribute() is not { IsError: true } select (Type: type, Method: method, Signature: $"({string.Join(", ", method.GetParameters().Select(p => Describe(p.ParameterType)))})"); @@ -250,6 +277,13 @@ public async Task Later(T value) return value; } + /// A stream whose first element is 's task, the second following it at once. + public async IAsyncEnumerable Stream(T first, T second) + { + yield return await First(first).ConfigureAwait(false); + yield return second; + } + public T Seen(T value) { Called = true; diff --git a/test/CodoMetis.TypeKit.Tests/NullDelegateTests.cs b/test/CodoMetis.TypeKit.Tests/NullDelegateTests.cs index 694c9db5..9a1ef06a 100644 --- a/test/CodoMetis.TypeKit.Tests/NullDelegateTests.cs +++ b/test/CodoMetis.TypeKit.Tests/NullDelegateTests.cs @@ -115,6 +115,16 @@ public sealed class NullDelegateTests [() => new[] { 1 }.TraverseAsync(Null>>>()), () => Array.Empty().TraverseAsync(Null>>>())], ["Result.TraverseAsync(IEnumerable, Func>>, CancellationToken): selector"] = [() => new[] { 1 }.TraverseAsync(Null>>>(), CancellationToken.None), () => Array.Empty().TraverseAsync(Null>>>(), CancellationToken.None)], + ["SequenceOptionExtensions.Choose(IAsyncEnumerable, Func>): selector"] = + [() => new[] { 1 }.ToAsyncEnumerable().Choose(Null>>()), () => Array.Empty().ToAsyncEnumerable().Choose(Null>>())], + ["Result.TraverseAsync(IAsyncEnumerable, Func>>): selector"] = + [() => new[] { 1 }.ToAsyncEnumerable().TraverseAsync(Null>>>()), () => Array.Empty().ToAsyncEnumerable().TraverseAsync(Null>>>())], + ["Result.TraverseAsync(IAsyncEnumerable, Func>>, CancellationToken): selector"] = + [() => new[] { 1 }.ToAsyncEnumerable().TraverseAsync(Null>>>(), CancellationToken.None), () => Array.Empty().ToAsyncEnumerable().TraverseAsync(Null>>>(), CancellationToken.None)], + ["Result.TraverseAsync(IAsyncEnumerable, Func>>): selector"] = + [() => new[] { 1 }.ToAsyncEnumerable().TraverseAsync(Null>>>()), () => Array.Empty().ToAsyncEnumerable().TraverseAsync(Null>>>())], + ["Result.TraverseAsync(IAsyncEnumerable, Func>>, CancellationToken): selector"] = + [() => new[] { 1 }.ToAsyncEnumerable().TraverseAsync(Null>>>(), CancellationToken.None), () => Array.Empty().ToAsyncEnumerable().TraverseAsync(Null>>>(), CancellationToken.None)], ["Result.TraverseAsync(Option, Func>>): selector"] = [() => Some.TraverseAsync(Null>>>()), () => None.TraverseAsync(Null>>>())], ["Result.TraverseAsync(Option, Func>>, CancellationToken): selector"] = diff --git a/test/CodoMetis.TypeKit.Tests/NullTaskTests.cs b/test/CodoMetis.TypeKit.Tests/NullTaskTests.cs index c1c10dfe..5407e0da 100644 --- a/test/CodoMetis.TypeKit.Tests/NullTaskTests.cs +++ b/test/CodoMetis.TypeKit.Tests/NullTaskTests.cs @@ -22,7 +22,11 @@ public sealed class NullTaskTests ["TraverseAsync(selector, token)"] = ("TraverseAsync", "selector", () => Elements.TraverseAsync((_, _) => (Task>)null!, CancellationToken.None)), ["command TraverseAsync(selector)"] = ("TraverseAsync", "selector", () => Elements.TraverseAsync(_ => (Task>)null!)), ["command TraverseAsync(selector, token)"] = ("TraverseAsync", "selector", () => Elements.TraverseAsync((_, _) => (Task>)null!, CancellationToken.None)), - ["Option.TraverseAsync(selector)"] = ("TraverseAsync", "selector", () => Some.TraverseAsync(_ => (Task>)null!)), + ["stream TraverseAsync(selector)"] = ("TraverseAsync", "selector", () => Elements.ToAsyncEnumerable().TraverseAsync(_ => (Task>)null!)), + ["stream TraverseAsync(selector, token)"] = ("TraverseAsync", "selector", () => Elements.ToAsyncEnumerable().TraverseAsync((_, _) => (Task>)null!, CancellationToken.None)), + ["stream command TraverseAsync(selector)"] = ("TraverseAsync", "selector", () => Elements.ToAsyncEnumerable().TraverseAsync(_ => (Task>)null!)), + ["stream command TraverseAsync(selector, token)"] = ("TraverseAsync", "selector", () => Elements.ToAsyncEnumerable().TraverseAsync((_, _) => (Task>)null!, CancellationToken.None)), + ["Option.TraverseAsync(selector)"] = ("TraverseAsync", "selector", () => Some.TraverseAsync(_ => (Task>)null!)), ["Option.TraverseAsync(selector, token)"] = ("TraverseAsync", "selector", () => Some.TraverseAsync((_, _) => (Task>)null!, CancellationToken.None)), ["Patch.TraverseAsync(selector)"] = ("TraverseAsync", "selector", () => Patch.Value(1).TraverseAsync(_ => (Task>)null!)), ["Patch.TraverseAsync(selector, token)"] = ("TraverseAsync", "selector", () => Patch.Value(1).TraverseAsync((_, _) => (Task>)null!, CancellationToken.None)), diff --git a/test/CodoMetis.TypeKit.Tests/UninitializedResultTests.cs b/test/CodoMetis.TypeKit.Tests/UninitializedResultTests.cs index d4dd31b5..0f334910 100644 --- a/test/CodoMetis.TypeKit.Tests/UninitializedResultTests.cs +++ b/test/CodoMetis.TypeKit.Tests/UninitializedResultTests.cs @@ -137,6 +137,12 @@ public sealed class UninitializedResultTests ["TraverseAsync(IEnumerable, Func>>, CancellationToken)"] = () => new[] { 1 }.TraverseAsync((_, _) => Task.FromResult(default(Result)), CancellationToken.None), ["Sequence(IEnumerable>)"] = () => Task.FromResult(new[] { Value, default }.Sequence()), ["Sequence(IEnumerable>)"] = () => Task.FromResult(new[] { Result.Success(), default }.Sequence()), + ["TraverseAsync(IAsyncEnumerable, Func>>)"] = () => new[] { 1 }.ToAsyncEnumerable().TraverseAsync(_ => Task.FromResult(default(Result))), + ["TraverseAsync(IAsyncEnumerable, Func>>, CancellationToken)"] = () => new[] { 1 }.ToAsyncEnumerable().TraverseAsync((_, _) => Task.FromResult(default(Result)), CancellationToken.None), + ["TraverseAsync(IAsyncEnumerable, Func>>)"] = () => new[] { 1 }.ToAsyncEnumerable().TraverseAsync(_ => Task.FromResult(default(Result))), + ["TraverseAsync(IAsyncEnumerable, Func>>, CancellationToken)"] = () => new[] { 1 }.ToAsyncEnumerable().TraverseAsync((_, _) => Task.FromResult(default(Result)), CancellationToken.None), + ["SequenceAsync(IAsyncEnumerable>, CancellationToken)"] = () => new[] { Value, default }.ToAsyncEnumerable().SequenceAsync(), + ["SequenceAsync(IAsyncEnumerable>, CancellationToken)"] = () => new[] { Result.Success(), default }.ToAsyncEnumerable().SequenceAsync(), }; public static TheoryData ExtensionMemberNames => [.. ExtensionMembers.Keys]; diff --git a/test/oracles/claims.json b/test/oracles/claims.json index 83cb6a4d..fee6749a 100644 --- a/test/oracles/claims.json +++ b/test/oracles/claims.json @@ -2200,5 +2200,77 @@ "heldBy": [ "NotWireTypeTests" ] + }, + { + "id": "S01-choose-reads-a-stream", + "source": "src/CodoMetis.TypeKit/README.md#find-one-in-a-collection", + "quote": "Both read an async stream too, as its elements arrive.", + "claim": "Both read an async stream too, as its elements arrive.", + "oracle": "claim", + "surfaces": [ + "Option.Choose", + "SequenceOptionExtensions.Choose" + ], + "heldBy": [ + "AsyncStreamTests" + ] + }, + { + "id": "S01-dispose-however-they-stop", + "source": "src/CodoMetis.TypeKit/README.md#chain-steps", + "quote": "All three read the stream in order, and dispose its enumerator however they stop: at an error, at a `break`, or when your selector throws.", + "claim": "All three read the stream in order, and dispose its enumerator however they stop: at an error, at a `break`, or when your selector throws.", + "oracle": "claim", + "surfaces": [ + "Option.Choose", + "SequenceOptionExtensions.Choose", + "Result.SequenceAsync", + "Result.TraverseAsync" + ], + "heldBy": [ + "AsyncStreamTests" + ] + }, + { + "id": "S01-stop-at-the-first-error", + "source": "src/CodoMetis.TypeKit/README.md#chain-steps", + "quote": "`SequenceAsync` and `TraverseAsync` stop at the first error and read nothing after it.", + "claim": "`SequenceAsync` and `TraverseAsync` stop at the first error and read nothing after it.", + "oracle": "claim", + "surfaces": [ + "Result.SequenceAsync", + "Result.TraverseAsync" + ], + "heldBy": [ + "AsyncStreamTests" + ] + }, + { + "id": "S01-token-reaches-the-stream", + "source": "src/CodoMetis.TypeKit/README.md#chain-steps", + "quote": "The token you give `SequenceAsync` or `TraverseAsync` reaches the stream.", + "claim": "The token you give `SequenceAsync` or `TraverseAsync` reaches the stream.", + "oracle": "claim", + "surfaces": [ + "Result.SequenceAsync", + "Result.TraverseAsync" + ], + "heldBy": [ + "AsyncStreamTests" + ] + }, + { + "id": "S01-with-cancellation-for-choose", + "source": "src/CodoMetis.TypeKit/README.md#chain-steps", + "quote": "For `Choose`, pass yours to `WithCancellation` on what it returns.", + "claim": "For `Choose`, pass yours to `WithCancellation` on what it returns.", + "oracle": "claim", + "surfaces": [ + "Option.Choose", + "SequenceOptionExtensions.Choose" + ], + "heldBy": [ + "AsyncStreamTests" + ] } ] diff --git a/test/oracles/surface-waivers.json b/test/oracles/surface-waivers.json index 7b89a2d0..f035ce26 100644 --- a/test/oracles/surface-waivers.json +++ b/test/oracles/surface-waivers.json @@ -19,7 +19,6 @@ "Option.*": "unread: the document that promises its behaviour has not been read for claims yet (#98); the core tests hold it meanwhile", "Result.*": "unread: the document that promises its behaviour has not been read for claims yet (#98); the core tests hold it meanwhile", "Patch.*": "unread: the document that promises its behaviour has not been read for claims yet (#98); the core tests hold it meanwhile", - "SequenceOptionExtensions.Choose": "unread: the document that promises its behaviour has not been read for claims yet (#98); the core tests hold it meanwhile", "ValueTypeOptionExtensions.OrNull": "unread: the document that promises its behaviour has not been read for claims yet (#98); the core tests hold it meanwhile", "GuidValueExtensions.New": "unread: the document that promises its behaviour has not been read for claims yet (#98); the core tests hold it meanwhile", "TypeKitSerilogExtensions.WithTypeKit": "unread: the document that promises its behaviour has not been read for claims yet (#98); the core tests hold it meanwhile",