Add TaskSeq.takeWhile, takeWhileAsync, takeWhileInclusive, takeWhileInclusiveAsync

README.md

@@ -52,7 +52,7 @@ Since the introduction of `task` in F# the call for a native implementation of _
 
 ### Module functions
 
-As with `seq` and `Seq`, this library comes with a bunch of well-known collection functions, like `TaskSeq.empty`, `isEmpty` or `TaskSeq.map`, `iter`, `collect`, `fold` and `TaskSeq.find`, `pick`, `choose`, `filter`. Where applicable, these come with async variants, like `TaskSeq.mapAsync` `iterAsync`, `collectAsync`, `foldAsync` and `TaskSeq.findAsync`, `pickAsync`, `chooseAsync`, `filterAsync`, which allows the applied function to be asynchronous.
+As with `seq` and `Seq`, this library comes with a bunch of well-known collection functions, like `TaskSeq.empty`, `isEmpty` or `TaskSeq.map`, `iter`, `collect`, `fold` and `TaskSeq.find`, `pick`, `choose`, `filter`, `takeWhile`. Where applicable, these come with async variants, like `TaskSeq.mapAsync` `iterAsync`, `collectAsync`, `foldAsync` and `TaskSeq.findAsync`, `pickAsync`, `chooseAsync`, `filterAsync`, `takeWhileAsync` which allows the applied function to be asynchronous.
 
 [See below](#current-set-of-taskseq-utility-functions) for a full list of currently implemented functions and their variants.
 
@@ -279,6 +279,8 @@ The following is the progress report:
 | ✅ [#90][] | `singleton`        | `singleton`     |                      | |
 |                  | `skip`             | `skip`          |                      | |
 |                  | `skipWhile`        | `skipWhile`     | `skipWhileAsync`     | |
+|                  |                    |                 | `skipWhileInclusive` | |
+|                  |                    |                 | `skipWhileInclusiveAsync` | |
 | ❓         | `sort`             |                 |                      | [note #1](#note1 "These functions require a form of pre-materializing through 'TaskSeq.cache', similar to the approach taken in the corresponding 'Seq' functions. It doesn't make much sense to have a cached async sequence. However, 'AsyncSeq' does implement these, so we'll probably do so eventually as well.") |
 | ❓         | `sortBy`           |                 |                      | [note #1](#note1 "These functions require a form of pre-materializing through 'TaskSeq.cache', similar to the approach taken in the corresponding 'Seq' functions. It doesn't make much sense to have a cached async sequence. However, 'AsyncSeq' does implement these, so we'll probably do so eventually as well.") |
 | ❓         | `sortByAscending`  |                 |                      | [note #1](#note1 "These functions require a form of pre-materializing through 'TaskSeq.cache', similar to the approach taken in the corresponding 'Seq' functions. It doesn't make much sense to have a cached async sequence. However, 'AsyncSeq' does implement these, so we'll probably do so eventually as well.") |
@@ -289,7 +291,9 @@ The following is the progress report:
 |                  | `sumBy`            | `sumBy`         | `sumByAsync`         | |
 | ✅ [#76][] | `tail`             | `tail`          |                      | |
 |                  | `take`             | `take`          |                      | |
-|                  | `takeWhile`        | `takeWhile`     | `takeWhileAsync`     | |
+| ✅ [#126][]| `takeWhile`        | `takeWhile`     | `takeWhileAsync`     | |
+| ✅ [#126][]|                    |                 | `takeWhileInclusive` | |
+| ✅ [#126][]|                    |                 | `takeWhileInclusiveAsync`| |
 | ✅ [#2][]  | `toArray`          | `toArray`       | `toArrayAsync`       | |
 | ✅ [#2][]  |                    | `toIList`       | `toIListAsync`       | |
 | ✅ [#2][]  | `toList`           | `toList`        | `toListAsync`        | |
@@ -545,6 +549,7 @@ module TaskSeq =
 [#82]: https://github.com/fsprojects/FSharp.Control.TaskSeq/pull/82
 [#83]: https://github.com/fsprojects/FSharp.Control.TaskSeq/pull/83
 [#90]: https://github.com/fsprojects/FSharp.Control.TaskSeq/pull/90
+[#126]: https://github.com/fsprojects/FSharp.Control.TaskSeq/pull/126
 
 [issues]: https://github.com/fsprojects/FSharp.Control.TaskSeq/issues
 [nuget]: https://www.nuget.org/packages/FSharp.Control.TaskSeq/

release-notes.txt

@@ -1,5 +1,7 @@
 
 Release notes:
+0.4.x (unreleased)
+    - adds TaskSeq.takeWhile, takeWhileAsync, takeWhileInclusive, takeWhileInclusiveAsync, #126 (by @bartelink)
 
 0.3.0
     - internal renames, improved doc comments, signature files for complex types, hide internal-only types, fixes #112.

src/FSharp.Control.TaskSeq.Test/FSharp.Control.TaskSeq.Test.fsproj

@@ -2,9 +2,6 @@
 
   <PropertyGroup>
     <TargetFramework>net6.0</TargetFramework>
-
-    <IsPackable>false</IsPackable>
-    <GenerateProgramFile>false</GenerateProgramFile>
   </PropertyGroup>
 
   <ItemGroup>
@@ -38,6 +35,7 @@
     <Compile Include="TaskSeq.OfXXX.Tests.fs" />
     <Compile Include="TaskSeq.Pick.Tests.fs" />
     <Compile Include="TaskSeq.Singleton.Tests.fs" />
+    <Compile Include="TaskSeq.TakeWhile.Tests.fs" />
     <Compile Include="TaskSeq.Tail.Tests.fs" />
     <Compile Include="TaskSeq.ToXXX.Tests.fs" />
     <Compile Include="TaskSeq.Zip.Tests.fs" />

src/FSharp.Control.TaskSeq.Test/TaskSeq.TakeWhile.Tests.fs

@@ -0,0 +1,258 @@
+module TaskSeq.Tests.TakeWhile
+
+open System
+open Xunit
+open FsUnit.Xunit
+open FsToolkit.ErrorHandling
+
+open FSharp.Control
+
+//
+// TaskSeq.takeWhile
+// TaskSeq.takeWhileAsync
+// TaskSeq.takeWhileInclusive
+// TaskSeq.takeWhileInclusiveAsync
+//
+
+[<AutoOpen>]
+module With =
+    /// The only real difference in semantics between the base and the *Inclusive variant lies in whether the final item is returned.
+    /// NOTE the semantics are very clear on only propagating a single failing item in the inclusive case.
+    let getFunction inclusive isAsync =
+        match inclusive, isAsync with
+        | false, false -> TaskSeq.takeWhile
+        | false, true -> fun pred -> TaskSeq.takeWhileAsync (pred >> Task.fromResult)
+        | true, false -> TaskSeq.takeWhileInclusive
+        | true, true -> fun pred -> TaskSeq.takeWhileInclusiveAsync (pred >> Task.fromResult)
+
+    /// adds '@' to each number and concatenates the chars before calling 'should equal'
+    let verifyAsString expected =
+        TaskSeq.map char
+        >> TaskSeq.map ((+) '@')
+        >> TaskSeq.toArrayAsync
+        >> Task.map (String >> should equal expected)
+
+    /// This is the base condition as one would expect in actual code
+    let inline cond x = x <> 6
+
+    /// For each of the tests below, we add a guard that will trigger if the predicate is passed items known to be beyond the
+    /// first failing item in the known sequence (which is 1..10)
+    let inline condWithGuard x =
+        let res = cond x
+
+        if x > 6 then
+            failwith "Test sequence should not be enumerated beyond the first item failing the predicate"
+
+        res
+
+module EmptySeq =
+    [<Theory; ClassData(typeof<TestEmptyVariants>)>]
+    let ``TaskSeq-takeWhile+A has no effect`` variant = task {
+        do! Gen.getEmptyVariant variant
+            |> TaskSeq.takeWhile ((=) 12)
+            |> verifyEmpty
+
+        do! Gen.getEmptyVariant variant
+            |> TaskSeq.takeWhileAsync ((=) 12 >> Task.fromResult)
+            |> verifyEmpty
+    }
+
+    [<Theory; ClassData(typeof<TestEmptyVariants>)>]
+    let ``TaskSeq-takeWhileInclusive+A has no effect`` variant = task {
+        do! Gen.getEmptyVariant variant
+            |> TaskSeq.takeWhileInclusive ((=) 12)
+            |> verifyEmpty
+
+        do! Gen.getEmptyVariant variant
+            |> TaskSeq.takeWhileInclusiveAsync ((=) 12 >> Task.fromResult)
+            |> verifyEmpty
+    }
+
+module Immutable =
+
+    [<Theory; ClassData(typeof<TestImmTaskSeq>)>]
+    let ``TaskSeq-takeWhile+A filters correctly`` variant = task {
+        do!
+            Gen.getSeqImmutable variant
+            |> TaskSeq.takeWhile condWithGuard
+            |> verifyAsString "ABCDE"
+
+        do!
+            Gen.getSeqImmutable variant
+            |> TaskSeq.takeWhileAsync (fun x -> task { return condWithGuard x })
+            |> verifyAsString "ABCDE"
+    }
+
+    [<Theory; ClassData(typeof<TestImmTaskSeq>)>]
+    let ``TaskSeq-takeWhile+A does not pick first item when false`` variant = task {
+        do!
+            Gen.getSeqImmutable variant
+            |> TaskSeq.takeWhile ((=) 0)
+            |> verifyAsString ""
+
+        do!
+            Gen.getSeqImmutable variant
+            |> TaskSeq.takeWhileAsync ((=) 0 >> Task.fromResult)
+            |> verifyAsString ""
+    }
+
+    [<Theory; ClassData(typeof<TestImmTaskSeq>)>]
+    let ``TaskSeq-takeWhileInclusive+A filters correctly`` variant = task {
+        do!
+            Gen.getSeqImmutable variant
+            |> TaskSeq.takeWhileInclusive condWithGuard
+            |> verifyAsString "ABCDEF"
+
+        do!
+            Gen.getSeqImmutable variant
+            |> TaskSeq.takeWhileInclusiveAsync (fun x -> task { return condWithGuard x })
+            |> verifyAsString "ABCDEF"
+    }
+
+    [<Theory; ClassData(typeof<TestImmTaskSeq>)>]
+    let ``TaskSeq-takeWhileInclusive+A always pick at least the first item`` variant = task {
+        do!
+            Gen.getSeqImmutable variant
+            |> TaskSeq.takeWhileInclusive ((=) 0)
+            |> verifyAsString "A"
+
+        do!
+            Gen.getSeqImmutable variant
+            |> TaskSeq.takeWhileInclusiveAsync ((=) 0 >> Task.fromResult)
+            |> verifyAsString "A"
+    }
+
+module SideEffects =
+    [<Theory; ClassData(typeof<TestSideEffectTaskSeq>)>]
+    let ``TaskSeq-takeWhile filters correctly`` variant =
+        Gen.getSeqWithSideEffect variant
+        |> TaskSeq.takeWhile condWithGuard
+        |> verifyAsString "ABCDE"
+
+    [<Theory; ClassData(typeof<TestSideEffectTaskSeq>)>]
+    let ``TaskSeq-takeWhileAsync filters correctly`` variant =
+        Gen.getSeqWithSideEffect variant
+        |> TaskSeq.takeWhileAsync (fun x -> task { return condWithGuard x })
+        |> verifyAsString "ABCDE"
+
+    [<Theory>]
+    [<InlineData(false, false)>]
+    [<InlineData(false, true)>]
+    [<InlineData(true, false)>]
+    [<InlineData(true, true)>]
+    let ``TaskSeq-takeWhileXXX prove it does not read beyond the failing yield`` (inclusive, isAsync) = task {
+        let mutable x = 42 // for this test, the potential mutation should not actually occur
+        let functionToTest = getFunction inclusive isAsync ((=) 42)
+
+        let items = taskSeq {
+            yield x // Always passes the test; always returned
+            yield x * 2 // the failing item (which will also be yielded in the result when using *Inclusive)
+            x <- x + 1 // we are proving we never get here
+        }
+
+        let expected = if inclusive then [| 42; 84 |] else [| 42 |]
+
+        let! first = items |> functionToTest |> TaskSeq.toArrayAsync
+        let! repeat = items |> functionToTest |> TaskSeq.toArrayAsync
+
+        first |> should equal expected
+        repeat |> should equal expected
+        x |> should equal 42
+    }
+
+    [<Theory>]
+    [<InlineData(false, false)>]
+    [<InlineData(false, true)>]
+    [<InlineData(true, false)>]
+    [<InlineData(true, true)>]
+    let ``TaskSeq-takeWhileXXX prove side effects are executed`` (inclusive, isAsync) = task {
+        let mutable x = 41
+        let functionToTest = getFunction inclusive isAsync ((>) 50)
+
+        let items = taskSeq {
+            x <- x + 1
+            yield x
+            x <- x + 2
+            yield x * 2
+            x <- x + 200 // as previously proven, we should not trigger this
+        }
+
+        let expectedFirst = if inclusive then [| 42; 44 * 2 |] else [| 42 |]
+        let expectedRepeat = if inclusive then [| 45; 47 * 2 |] else [| 45 |]
+
+        let! first = items |> functionToTest |> TaskSeq.toArrayAsync
+        x |> should equal 44
+        let! repeat = items |> functionToTest |> TaskSeq.toArrayAsync
+        x |> should equal 47
+
+        first |> should equal expectedFirst
+        repeat |> should equal expectedRepeat
+    }
+
+    [<Theory; ClassData(typeof<TestSideEffectTaskSeq>)>]
+    let ``TaskSeq-takeWhile consumes the prefix of a longer sequence, with mutation`` variant = task {
+        let ts = Gen.getSeqWithSideEffect variant
+
+        let! first =
+            TaskSeq.takeWhile (fun x -> x < 5) ts
+            |> TaskSeq.toArrayAsync
+
+        let expected = [| 1..4 |]
+        first |> should equal expected
+
+        // side effect, reiterating causes it to resume from where we left it (minus the failing item)
+        let! repeat =
+            TaskSeq.takeWhile (fun x -> x < 5) ts
+            |> TaskSeq.toArrayAsync
+
+        repeat |> should not' (equal expected)
+    }
+
+    [<Theory; ClassData(typeof<TestSideEffectTaskSeq>)>]
+    let ``TaskSeq-takeWhileInclusiveAsync consumes the prefix for a longer sequence, with mutation`` variant = task {
+        let ts = Gen.getSeqWithSideEffect variant
+
+        let! first =
+            TaskSeq.takeWhileInclusiveAsync (fun x -> task { return x < 5 }) ts
+            |> TaskSeq.toArrayAsync
+
+        let expected = [| 1..5 |]
+        first |> should equal expected
+
+        // side effect, reiterating causes it to resume from where we left it (minus the failing item)
+        let! repeat =
+            TaskSeq.takeWhileInclusiveAsync (fun x -> task { return x < 5 }) ts
+            |> TaskSeq.toArrayAsync
+
+        repeat |> should not' (equal expected)
+    }
+
+module Other =
+    [<Theory>]
+    [<InlineData(false, false)>]
+    [<InlineData(false, true)>]
+    [<InlineData(true, false)>]
+    [<InlineData(true, true)>]
+    let ``TaskSeq-takeWhileXXX exclude all items after predicate fails`` (inclusive, isAsync) =
+        let functionToTest = With.getFunction inclusive isAsync
+
+        [ 1; 2; 2; 3; 3; 2; 1 ]
+        |> TaskSeq.ofSeq
+        |> functionToTest (fun x -> x <= 2)
+        |> verifyAsString (if inclusive then "ABBC" else "ABB")
+
+    [<Theory>]
+    [<InlineData(false, false)>]
+    [<InlineData(false, true)>]
+    [<InlineData(true, false)>]
+    [<InlineData(true, true)>]
+    let ``TaskSeq-takeWhileXXX stops consuming after predicate fails`` (inclusive, isAsync) =
+        let functionToTest = With.getFunction inclusive isAsync
+
+        seq {
+            yield! [ 1; 2; 2; 3; 3 ]
+            yield failwith "Too far"
+        }
+        |> TaskSeq.ofSeq
+        |> functionToTest (fun x -> x <= 2)
+        |> verifyAsString (if inclusive then "ABBC" else "ABB")

src/FSharp.Control.TaskSeq/TaskSeq.fs

@@ -253,6 +253,10 @@ module TaskSeq =
     let chooseAsync chooser source = Internal.choose (TryPickAsync chooser) source
     let filter predicate source = Internal.filter (Predicate predicate) source
     let filterAsync predicate source = Internal.filter (PredicateAsync predicate) source
+    let takeWhile predicate source = Internal.takeWhile Exclusive (Predicate predicate) source
+    let takeWhileAsync predicate source = Internal.takeWhile Exclusive (PredicateAsync predicate) source
+    let takeWhileInclusive predicate source = Internal.takeWhile Inclusive (Predicate predicate) source
+    let takeWhileInclusiveAsync predicate source = Internal.takeWhile Inclusive (PredicateAsync predicate) source
     let tryPick chooser source = Internal.tryPick (TryPick chooser) source
     let tryPickAsync chooser source = Internal.tryPick (TryPickAsync chooser) source
     let tryFind predicate source = Internal.tryFind (Predicate predicate) source

src/FSharp.Control.TaskSeq/TaskSeq.fsi

@@ -365,6 +365,36 @@ module TaskSeq =
     /// </summary>
     val filter: predicate: ('T -> bool) -> source: taskSeq<'T> -> taskSeq<'T>
 
+    /// <summary>
+    /// Yields items from the source while the <paramref name="predicate" /> function returns <see cref="true" />.
+    /// The first <see cref="false" /> result concludes consumption of the source.
+    /// If <paramref name="predicate" /> is asynchronous, consider using <see cref="TaskSeq.takeWhileAsync" />.
+    /// </summary>
+    val takeWhile: predicate: ('T -> bool) -> source: taskSeq<'T> -> taskSeq<'T>
+
+    /// <summary>
+    /// Yields items from the source while the <paramref name="predicate" /> asynchronous function returns <see cref="true" />.
+    /// The first <see cref="false" /> result concludes consumption of the source.
+    /// If <paramref name="predicate" /> does not need to be asynchronous, consider using <see cref="TaskSeq.takeWhile" />.
+    /// </summary>
+    val takeWhileAsync: predicate: ('T -> #Task<bool>) -> source: taskSeq<'T> -> taskSeq<'T>
+
+    /// <summary>
+    /// Yields items from the source while the <paramref name="predicate" /> function returns <see cref="true" />.
+    /// The first <see cref="false" /> result concludes consumption of the source, but is included in the result.
+    /// If <paramref name="predicate" /> is asynchronous, consider using <see cref="TaskSeq.takeWhileInclusiveAsync" />.
+    /// If the final item is not desired, consider using <see cref="TaskSeq.takeWhile" />.
+    /// </summary>
+    val takeWhileInclusive: predicate: ('T -> bool) -> source: taskSeq<'T> -> taskSeq<'T>
+
+    /// <summary>
+    /// Yields items from the source while the <paramref name="predicate" /> asynchronous function returns <see cref="true" />.
+    /// The first <see cref="false" /> result concludes consumption of the source, but is included in the result.
+    /// If <paramref name="predicate" /> does not need to be asynchronous, consider using <see cref="TaskSeq.takeWhileInclusive" />.
+    /// If the final item is not desired, consider using <see cref="TaskSeq.takeWhileAsync" />.
+    /// </summary>
+    val takeWhileInclusiveAsync: predicate: ('T -> #Task<bool>) -> source: taskSeq<'T> -> taskSeq<'T>
+
     /// <summary>
     /// Returns a new collection containing only the elements of the collection
     /// for which the given asynchronous function <paramref name="predicate" /> returns <see cref="true" />.