Fix resumability of iteration over IAsyncEnumerable<'T>

src/FSharpy.TaskSeq.Test/FSharpy.TaskSeq.Test.fsproj

@@ -29,6 +29,8 @@
     <Compile Include="TaskSeq.OfXXX.Tests.fs" />
     <Compile Include="TaskSeq.Tests.Other.fs" />
     <Compile Include="TaskSeq.Tests.CE.fs" />
+    <Compile Include="TaskSeq.StateTransitionBug.Tests.CE.fs" />
+    <Compile Include="TaskSeq.StateTransitionBug-delayed.Tests.CE.fs" />
     <Compile Include="TaskSeq.PocTests.fs" />
     <Compile Include="TaskSeq.Realworld.fs" />
     <Compile Include="Program.fs" />

src/FSharpy.TaskSeq.Test/TaskSeq.Iter.Tests.fs

@@ -37,6 +37,18 @@ let ``TaskSeq-iter should go over all items`` () = task {
     sum |> should equal 55 // task-dummies started at 1
 }
 
+
+[<Fact>]
+let ``TaskSeq-iter multiple iterations over same sequence`` () = task {
+    let tq = createDummyTaskSeq 10
+    let mutable sum = 0
+    do! tq |> TaskSeq.iter (fun item -> sum <- sum + item)
+    do! tq |> TaskSeq.iter (fun item -> sum <- sum + item)
+    do! tq |> TaskSeq.iter (fun item -> sum <- sum + item)
+    do! tq |> TaskSeq.iter (fun item -> sum <- sum + item)
+    sum |> should equal 820 // task-dummies started at 1
+}
+
 [<Fact>]
 let ``TaskSeq-iteriAsync should go over all items`` () = task {
     let tq = createDummyTaskSeq 10

src/FSharpy.TaskSeq.Test/TaskSeq.Map.Tests.fs

@@ -6,12 +6,25 @@ open FsToolkit.ErrorHandling
 
 open FSharpy
 
+/// Asserts that a sequence contains the char values 'A'..'J'.
 let validateSequence sequence =
     sequence
     |> Seq.map string
     |> String.concat ""
     |> should equal "ABCDEFGHIJ"
 
+/// Validates for "ABCDEFGHIJ" char sequence, or any amount of char-value higher
+let validateSequenceWithOffset offset sequence =
+    let expected =
+        [ 'A' .. 'J' ]
+        |> List.map (int >> (+) offset >> char >> string)
+        |> String.concat ""
+
+    sequence
+    |> Seq.map string
+    |> String.concat ""
+    |> should equal expected
+
 [<Fact>]
 let ``TaskSeq-map maps in correct order`` () = task {
     let! sq =
@@ -72,6 +85,27 @@ let ``TaskSeq-mapAsync maps in correct order`` () = task {
     validateSequence sq
 }
 
+[<Fact>]
+let ``TaskSeq-mapAsync can map the same sequence multiple times`` () = task {
+    let mapAndCache =
+        TaskSeq.mapAsync (fun item -> task { return char (item + 64) })
+        >> TaskSeq.toSeqCachedAsync
+
+    let ts = createDummyDirectTaskSeq 10
+
+    let! result1 = mapAndCache ts
+    let! result2 = mapAndCache ts
+    let! result3 = mapAndCache ts
+    let! result4 = mapAndCache ts
+    validateSequence result1
+
+    // each time we do GetAsyncEnumerator(), and go through the whole sequence,
+    // the whole sequence gets re-evaluated, causing our +1 side-effect to run again.
+    validateSequenceWithOffset 10 result2 // the mutable is 10 higher
+    validateSequenceWithOffset 20 result3 // again
+    validateSequenceWithOffset 30 result4 // again
+}
+
 [<Fact>]
 let ``TaskSeq-mapiAsync maps in correct order`` () = task {
     let! sq =

src/FSharpy.TaskSeq.Test/TaskSeq.Realworld.fs

@@ -22,8 +22,14 @@ type AsyncBufferedReader(output: ITestOutputHelper, data, blockSize) =
 
     interface IAsyncEnumerable<byte[]> with
         member reader.GetAsyncEnumerator(ct) =
-            output.WriteLine $"Cloning!! Current: {current}, lastPos: {lastPos}"
-            reader.MemberwiseClone() :?> IAsyncEnumerator<_>
+            { new IAsyncEnumerator<_> with
+                member this.Current = (reader :> IAsyncEnumerator<_>).Current
+                member this.MoveNextAsync() = (reader :> IAsyncEnumerator<_>).MoveNextAsync()
+              interface IAsyncDisposable with
+                  member this.DisposeAsync() = ValueTask()
+            }
+    //output.WriteLine $"Cloning!! Current: {current}, lastPos: {lastPos}"
+    //reader.MemberwiseClone() :?> IAsyncEnumerator<_>
 
     interface IAsyncEnumerator<byte[]> with
         member _.Current =
@@ -39,6 +45,8 @@ type AsyncBufferedReader(output: ITestOutputHelper, data, blockSize) =
                 let! bytesRead = buffered.ReadAsync(mem, 0, mem.Length) // offset refers to offset in target buffer, not source
                 lastPos <- buffered.Position
 
+                let x: seq<Guid> = seq { 1 } |> Seq.cast
+
                 if bytesRead > 0 then
                     current <- ValueSome mem
                     return true
@@ -48,7 +56,6 @@ type AsyncBufferedReader(output: ITestOutputHelper, data, blockSize) =
             }
             |> Task.toValueTask
 
-    interface IAsyncDisposable with
         member _.DisposeAsync() =
             try
                 // this disposes of the mem stream
@@ -57,8 +64,44 @@ type AsyncBufferedReader(output: ITestOutputHelper, data, blockSize) =
                 // if the previous block raises, we should still try to get rid of the underlying stream
                 stream.DisposeAsync().AsTask().Wait()
 
+
 type ``Real world tests``(output: ITestOutputHelper) =
     [<Fact>]
+    let ``Reading a 10MB buffered IAsync through TaskSeq.toArray non-async should succeed`` () = task {
+        use reader = AsyncBufferedReader(output, Array.init 2048 byte, 256)
+        // unreadable error with 'use'
+        //use bla = seq { 1}
+        let expected = Array.init 256 byte |> Array.replicate 8
+        let results = reader |> TaskSeq.toArray
+
+        (results, expected)
+        ||> Array.iter2 (fun a b -> should equal a b)
+    }
+
+    [<Fact(Skip = "Broken test, faulty streaming test-implementation")>]
+    let ``Reading a user-code IAsync multiple times with TaskSeq.toArrayAsync should succeed`` () = task {
+        use reader = AsyncBufferedReader(output, Array.init 2048 byte, 256)
+        let expected = Array.init 256 byte |> Array.replicate 8
+        // read four times
+        let! results1 = reader |> TaskSeq.toArrayAsync
+        let! results2 = reader |> TaskSeq.toArrayAsync
+        let! results3 = reader |> TaskSeq.toArrayAsync
+        let! results4 = reader |> TaskSeq.toArrayAsync
+
+        (results1, expected)
+        ||> Array.iter2 (fun a b -> should equal a b)
+
+        (results2, expected)
+        ||> Array.iter2 (fun a b -> should equal a b)
+
+        (results3, expected)
+        ||> Array.iter2 (fun a b -> should equal a b)
+
+        (results4, expected)
+        ||> Array.iter2 (fun a b -> should equal a b)
+    }
+
+    [<Fact(Skip = "Broken test, faulty streaming test-implementation")>]
     let ``Reading a 10MB buffered IAsync stream from start to finish`` () = task {
         let mutable count = 0
         use reader = AsyncBufferedReader(output, Array.init 2048 byte, 256)
@@ -76,6 +119,8 @@ type ``Real world tests``(output: ITestOutputHelper) =
 
         // the following is extremely slow, which is why we just use F#'s comparison instead
         // Using this takes 67s, compared to 0.25s using normal F# comparison.
+        // reader |> TaskSeq.toArray |> should equal expected // VERY SLOW!!
+
         do! reader |> TaskSeq.iter (should equal expected)
         do! reader |> TaskSeq.iter ((=) expected >> (should be True))
         let! len = reader |> TaskSeq.mapi (fun i _ -> i + 1) |> TaskSeq.last
@@ -113,13 +158,17 @@ type ``Real world tests``(output: ITestOutputHelper) =
     }
 
 
+    // This test used to have the following, which has since been solved through #42
+    // please leave this test in, as it tests a case that's quite easily reached if we
+    // introduce mistakes in the resumable code.
+    //
     //System.InvalidOperationException: An attempt was made to transition a task to a final state when it had already completed.
     //   at <StartupCode$FSharpy-TaskSeq-Test>.$TaskSeq.Realworld.clo@58-4.MoveNext() in D:\Projects\OpenSource\Abel\TaskSeq\src\FSharpy.TaskSeq.Test\TaskSeq.Realworld.fs:line 77
     //   at Xunit.Sdk.TestInvoker`1.<>c__DisplayClass48_0.<<InvokeTestMethodAsync>b__1>d.MoveNext() in /_/src/xunit.execution/Sdk/Frameworks/Runners/TestInvoker.cs:line 264
     //--- End of stack trace from previous location ---
     //   at Xunit.Sdk.ExecutionTimer.AggregateAsync(Func`1 asyncAction) in /_/src/xunit.execution/Sdk/Frameworks/ExecutionTimer.cs:line 48
     //   at Xunit.Sdk.ExceptionAggregator.RunAsync(Func`1 code) in /_/src/xunit.core/Sdk/ExceptionAggregator.cs:line 90\
-    [<Fact(Skip = "Currently fails")>]
+    [<Fact>]
     let ``Reading a 1MB buffered IAsync stream from start to finish InvalidOperationException`` () = task {
         let mutable count = 0
         use reader = AsyncBufferedReader(output, Array.init 1_048_576 byte, 256)