Skip to content

Commit 45e1abd

Browse files
CopilotTheAngryByrd
Copilot
and
TheAngryByrd
authored
Add IAsyncEnumerable support to task computation expressions (#347)
* Initial plan * Add IAsyncEnumerable support to taskResult, taskOption, taskValidation, taskValueOption CEs Co-authored-by: TheAngryByrd <1490044+TheAngryByrd@users.noreply.github.com> * Add IAsyncEnumerable tests for backgroundTask CE variants Co-authored-by: TheAngryByrd <1490044+TheAngryByrd@users.noreply.github.com> --------- Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com> Co-authored-by: TheAngryByrd <1490044+TheAngryByrd@users.noreply.github.com>
1 parent ff1f64f commit 45e1abd

12 files changed

Lines changed: 721 additions & 0 deletions

‎src/FsToolkit.ErrorHandling/TaskOptionCE.fs‎

Lines changed: 80 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@ open Microsoft.FSharp.Core.CompilerServices.StateMachineHelpers
1414
open Microsoft.FSharp.Core.LanguagePrimitives.IntrinsicOperators
1515
open Microsoft.FSharp.Control
1616
open Microsoft.FSharp.Collections
17+
open System.Collections.Generic
1718

1819
/// Task<'T option>
1920
type TaskOption<'T> = Task<'T option>
@@ -197,6 +198,78 @@ type TaskOptionBuilderBase() =
197198
)
198199
)
199200

201+
member inline internal this.WhileAsync
202+
(
203+
[<InlineIfLambda>] condition: unit -> ValueTask<bool>,
204+
body: TaskOptionCode<'TOverall, unit>
205+
) : TaskOptionCode<'TOverall, unit> =
206+
let mutable condition_res = true
207+
208+
this.While(
209+
(fun () -> condition_res),
210+
TaskOptionCode<_, _>(fun sm ->
211+
if __useResumableCode then
212+
let mutable __stack_condition_fin = true
213+
let __stack_vtask = condition ()
214+
let mutable awaiter = __stack_vtask.GetAwaiter()
215+
216+
if not awaiter.IsCompleted then
217+
let __stack_yield_fin = ResumableCode.Yield().Invoke(&sm)
218+
__stack_condition_fin <- __stack_yield_fin
219+
220+
if not __stack_condition_fin then
221+
sm.Data.MethodBuilder.AwaitUnsafeOnCompleted(&awaiter, &sm)
222+
223+
if __stack_condition_fin then
224+
condition_res <- awaiter.GetResult()
225+
226+
if condition_res then body.Invoke(&sm) else true
227+
else
228+
false
229+
else
230+
let vtask = condition ()
231+
let mutable awaiter = vtask.GetAwaiter()
232+
233+
let cont =
234+
TaskOptionResumptionFunc<'TOverall>(fun sm ->
235+
condition_res <- awaiter.GetResult()
236+
237+
if condition_res then body.Invoke(&sm) else true
238+
)
239+
240+
if awaiter.IsCompleted then
241+
cont.Invoke(&sm)
242+
else
243+
sm.ResumptionDynamicInfo.ResumptionData <-
244+
(awaiter :> ICriticalNotifyCompletion)
245+
246+
sm.ResumptionDynamicInfo.ResumptionFunc <- cont
247+
false
248+
)
249+
)
250+
251+
/// <summary>Iterates over an IAsyncEnumerable sequence, running the body for each element.</summary>
252+
/// <remarks>
253+
/// The existence of this method permits the use of <c>for .. in ..</c> with <see cref="T:System.Collections.Generic.IAsyncEnumerable`1"/>
254+
/// in the <c>taskOption { ... }</c> computation expression syntax.
255+
/// If the body returns None, the iteration stops early.
256+
/// </remarks>
257+
/// <param name="source">The async sequence to enumerate.</param>
258+
/// <param name="body">A function to take an item from the sequence and create a TaskOption.</param>
259+
/// <returns>A TaskOption that iterates the sequence and runs the body for each element.</returns>
260+
member inline this.For
261+
(source: #IAsyncEnumerable<'T>, body: 'T -> TaskOptionCode<'TOverall, unit>)
262+
: TaskOptionCode<'TOverall, unit> =
263+
this.Using(
264+
source.GetAsyncEnumerator(CancellationToken.None),
265+
(fun (e: IAsyncEnumerator<'T>) ->
266+
this.WhileAsync(
267+
(fun () -> e.MoveNextAsync()),
268+
TaskOptionCode<_, _>(fun sm -> (body e.Current).Invoke(&sm))
269+
)
270+
)
271+
)
272+
200273
member inline this.Source(taskOption: TaskOption<'T>) : TaskOption<'T> = taskOption
201274

202275
type TaskOptionBuilder() =
@@ -563,6 +636,13 @@ module TaskOptionCEExtensionsHighPriority =
563636

564637
member inline _.Source(s: #seq<_>) = s
565638

639+
[<AutoOpen>]
640+
module TaskOptionCEExtensionsHighPriorityAsyncEnumerable =
641+
642+
type TaskOptionBuilderBase with
643+
644+
member inline _.Source(source: #IAsyncEnumerable<_>) = source
645+
566646
[<AutoOpen>]
567647
module TaskOptionCEExtensionsMediumPriority =
568648

‎src/FsToolkit.ErrorHandling/TaskResultCE.fs‎

Lines changed: 80 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@ open Microsoft.FSharp.Core.CompilerServices.StateMachineHelpers
1414
open Microsoft.FSharp.Core.LanguagePrimitives.IntrinsicOperators
1515
open Microsoft.FSharp.Control
1616
open Microsoft.FSharp.Collections
17+
open System.Collections.Generic
1718

1819
/// Task<Result<'T, 'Error>>
1920
type TaskResult<'T, 'Error> = Task<Result<'T, 'Error>>
@@ -204,6 +205,78 @@ type TaskResultBuilderBase() =
204205
)
205206

206207

208+
member inline internal this.WhileAsync
209+
(
210+
[<InlineIfLambda>] condition: unit -> ValueTask<bool>,
211+
body: TaskResultCode<'TOverall, 'Error, unit>
212+
) : TaskResultCode<'TOverall, 'Error, unit> =
213+
let mutable condition_res = true
214+
215+
this.While(
216+
(fun () -> condition_res),
217+
TaskResultCode<_, _, _>(fun sm ->
218+
if __useResumableCode then
219+
let mutable __stack_condition_fin = true
220+
let __stack_vtask = condition ()
221+
let mutable awaiter = __stack_vtask.GetAwaiter()
222+
223+
if not awaiter.IsCompleted then
224+
let __stack_yield_fin = ResumableCode.Yield().Invoke(&sm)
225+
__stack_condition_fin <- __stack_yield_fin
226+
227+
if not __stack_condition_fin then
228+
sm.Data.MethodBuilder.AwaitUnsafeOnCompleted(&awaiter, &sm)
229+
230+
if __stack_condition_fin then
231+
condition_res <- awaiter.GetResult()
232+
233+
if condition_res then body.Invoke(&sm) else true
234+
else
235+
false
236+
else
237+
let vtask = condition ()
238+
let mutable awaiter = vtask.GetAwaiter()
239+
240+
let cont =
241+
TaskResultResumptionFunc<'TOverall, 'Error>(fun sm ->
242+
condition_res <- awaiter.GetResult()
243+
244+
if condition_res then body.Invoke(&sm) else true
245+
)
246+
247+
if awaiter.IsCompleted then
248+
cont.Invoke(&sm)
249+
else
250+
sm.ResumptionDynamicInfo.ResumptionData <-
251+
(awaiter :> ICriticalNotifyCompletion)
252+
253+
sm.ResumptionDynamicInfo.ResumptionFunc <- cont
254+
false
255+
)
256+
)
257+
258+
/// <summary>Iterates over an IAsyncEnumerable sequence, running the body for each element.</summary>
259+
/// <remarks>
260+
/// The existence of this method permits the use of <c>for .. in ..</c> with <see cref="T:System.Collections.Generic.IAsyncEnumerable`1"/>
261+
/// in the <c>taskResult { ... }</c> computation expression syntax.
262+
/// If the body returns an Error, the iteration stops early.
263+
/// </remarks>
264+
/// <param name="source">The async sequence to enumerate.</param>
265+
/// <param name="body">A function to take an item from the sequence and create a TaskResult.</param>
266+
/// <returns>A TaskResult that iterates the sequence and runs the body for each element.</returns>
267+
member inline this.For
268+
(source: #IAsyncEnumerable<'T>, body: 'T -> TaskResultCode<'TOverall, 'Error, unit>)
269+
: TaskResultCode<'TOverall, 'Error, unit> =
270+
this.Using(
271+
source.GetAsyncEnumerator(CancellationToken.None),
272+
(fun (e: IAsyncEnumerator<'T>) ->
273+
this.WhileAsync(
274+
(fun () -> e.MoveNextAsync()),
275+
TaskResultCode<_, _, _>(fun sm -> (body e.Current).Invoke(&sm))
276+
)
277+
)
278+
)
279+
207280
member inline this.Source(taskResult: TaskResult<'T, 'Error>) : TaskResult<'T, 'Error> =
208281
taskResult
209282

@@ -556,6 +629,13 @@ module TaskResultCEExtensionsHighPriority =
556629

557630
member inline _.Source(s: #seq<_>) = s
558631

632+
[<AutoOpen>]
633+
module TaskResultCEExtensionsHighPriorityAsyncEnumerable =
634+
635+
type TaskResultBuilderBase with
636+
637+
member inline _.Source(source: #IAsyncEnumerable<_>) = source
638+
559639
[<AutoOpen>]
560640
module TaskResultCEExtensionsMediumPriority =
561641

‎src/FsToolkit.ErrorHandling/TaskValidationCE.fs‎

Lines changed: 80 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@ open Microsoft.FSharp.Core.CompilerServices.StateMachineHelpers
1010
open Microsoft.FSharp.Core.LanguagePrimitives.IntrinsicOperators
1111
open Microsoft.FSharp.Control
1212
open Microsoft.FSharp.Collections
13+
open System.Collections.Generic
1314

1415
[<Struct; NoComparison; NoEquality>]
1516
type TaskValidationStateMachineData<'T, 'Error> =
@@ -198,6 +199,78 @@ type TaskValidationBuilderBase() =
198199
)
199200

200201

202+
member inline internal this.WhileAsync
203+
(
204+
[<InlineIfLambda>] condition: unit -> ValueTask<bool>,
205+
body: TaskValidationCode<'TOverall, 'Error, unit>
206+
) : TaskValidationCode<'TOverall, 'Error, unit> =
207+
let mutable condition_res = true
208+
209+
this.While(
210+
(fun () -> condition_res),
211+
TaskValidationCode<_, _, _>(fun sm ->
212+
if __useResumableCode then
213+
let mutable __stack_condition_fin = true
214+
let __stack_vtask = condition ()
215+
let mutable awaiter = __stack_vtask.GetAwaiter()
216+
217+
if not awaiter.IsCompleted then
218+
let __stack_yield_fin = ResumableCode.Yield().Invoke(&sm)
219+
__stack_condition_fin <- __stack_yield_fin
220+
221+
if not __stack_condition_fin then
222+
sm.Data.MethodBuilder.AwaitUnsafeOnCompleted(&awaiter, &sm)
223+
224+
if __stack_condition_fin then
225+
condition_res <- awaiter.GetResult()
226+
227+
if condition_res then body.Invoke(&sm) else true
228+
else
229+
false
230+
else
231+
let vtask = condition ()
232+
let mutable awaiter = vtask.GetAwaiter()
233+
234+
let cont =
235+
TaskValidationResumptionFunc<'TOverall, 'Error>(fun sm ->
236+
condition_res <- awaiter.GetResult()
237+
238+
if condition_res then body.Invoke(&sm) else true
239+
)
240+
241+
if awaiter.IsCompleted then
242+
cont.Invoke(&sm)
243+
else
244+
sm.ResumptionDynamicInfo.ResumptionData <-
245+
(awaiter :> ICriticalNotifyCompletion)
246+
247+
sm.ResumptionDynamicInfo.ResumptionFunc <- cont
248+
false
249+
)
250+
)
251+
252+
/// <summary>Iterates over an IAsyncEnumerable sequence, running the body for each element.</summary>
253+
/// <remarks>
254+
/// The existence of this method permits the use of <c>for .. in ..</c> with <see cref="T:System.Collections.Generic.IAsyncEnumerable`1"/>
255+
/// in the <c>taskValidation { ... }</c> computation expression syntax.
256+
/// If the body returns an Error, the iteration stops early.
257+
/// </remarks>
258+
/// <param name="source">The async sequence to enumerate.</param>
259+
/// <param name="body">A function to take an item from the sequence and create a TaskValidation.</param>
260+
/// <returns>A TaskValidation that iterates the sequence and runs the body for each element.</returns>
261+
member inline this.For
262+
(source: #IAsyncEnumerable<'T>, body: 'T -> TaskValidationCode<'TOverall, 'Error, unit>)
263+
: TaskValidationCode<'TOverall, 'Error, unit> =
264+
this.Using(
265+
source.GetAsyncEnumerator(CancellationToken.None),
266+
(fun (e: IAsyncEnumerator<'T>) ->
267+
this.WhileAsync(
268+
(fun () -> e.MoveNextAsync()),
269+
TaskValidationCode<_, _, _>(fun sm -> (body e.Current).Invoke(&sm))
270+
)
271+
)
272+
)
273+
201274
member inline this.Source
202275
(taskValidation: TaskValidation<'T, 'Error>)
203276
: TaskValidation<'T, 'Error> =
@@ -539,6 +612,13 @@ module TaskValidationCEExtensionsHighPriority =
539612

540613
member inline _.Source(s: #seq<_>) = s
541614

615+
[<AutoOpen>]
616+
module TaskValidationCEExtensionsHighPriorityAsyncEnumerable =
617+
618+
type TaskValidationBuilderBase with
619+
620+
member inline _.Source(source: #IAsyncEnumerable<_>) = source
621+
542622
[<AutoOpen>]
543623
module TaskValidationCEExtensionsMediumPriority =
544624

0 commit comments

Comments
 (0)