AsyncSeqSrc tests

Author: eulerfx

src/FSharp.Control.AsyncSeq/AsyncSeq.fs

@@ -25,7 +25,7 @@ type IAsyncEnumerable<'T> =
 type AsyncSeq<'T> = IAsyncEnumerable<'T>
 //    abstract GetEnumerator : unit -> IAsyncEnumerator<'T>
 
-type AsyncSeqSrc<'a> = private { mutable tl : TaskCompletionSource<('a * AsyncSeqSrc<'a>) option> }
+type AsyncSeqSrc<'a> = private { mutable tail : TaskCompletionSource<('a * AsyncSeqSrc<'a>) option> }
 
 [<AutoOpen>]
 module internal Utils = 
@@ -1369,23 +1369,29 @@ module AsyncSeq =
   module AsyncSeqSrcImpl =
 
     let create () : AsyncSeqSrc<'a> =
-      { tl = new TaskCompletionSource<_>() }
-
-    let put (a:'a) (s:AsyncSeqSrc<'a>) : unit =
-      let s' = create ()
-      s.tl.SetResult(Some(a, s'))
-      s.tl <- s'.tl
+      { tail = new TaskCompletionSource<_>() }
 
+    let put (a:'a) (s:AsyncSeqSrc<'a>) : unit =      
+      let newTail = create ()
+      let tail = s.tail
+      s.tail <- newTail.tail
+      tail.SetResult(Some(a, newTail))
+      
     let close (s:AsyncSeqSrc<'a>) : unit =
-      s.tl.SetResult(None)
+      s.tail.SetResult(None)
 
-    let rec toAsyncSeq (s:AsyncSeqSrc<'a>) : AsyncSeq<'a> = asyncSeq {
-      let! next = s.tl.Task |> Async.AwaitTask
-      match next with
-      | None -> ()
-      | Some (a,tl) ->
-        yield a
-        yield! toAsyncSeq tl }
+    let fail (ex:exn) (s:AsyncSeqSrc<'a>) : unit =
+      s.tail.SetException ex
+
+    let rec toAsyncSeq (s:AsyncSeqSrc<'a>) : AsyncSeq<'a> = 
+      let tail = s.tail
+      asyncSeq {
+        let! next = tail.Task |> Async.AwaitTask
+        match next with
+        | None -> ()
+        | Some (a,tl) ->
+          yield a
+          yield! toAsyncSeq tl }
 
 
   type private Group<'k, 'a> = { key : 'k ; src : AsyncSeqSrc<'a> }

tests/FSharp.Control.AsyncSeq.Tests/AsyncSeqTests.fs

@@ -28,6 +28,15 @@ let EQ (a:AsyncSeq<'a>) (b:AsyncSeq<'a>) =
     printfn "actual=%A" act
     false
 
+type Assert with  
+  /// Determines equality of two async sequences by convering them to lists, ignoring side-effects.
+  static member AreEqual (expected:AsyncSeq<'a>, actual:AsyncSeq<'a>) =
+    let exp = expected |> AsyncSeq.toList 
+    let act = actual |> AsyncSeq.toList 
+    let message = sprintf "expected=%A actual=%A" exp act
+    Assert.True((exp = act), message)
+    
+
 
 [<Test>]
 let ``AsyncSeq.toArray``() =  
@@ -1066,4 +1075,45 @@ let ``AsyncSeq.take should work``() =
   }
   let ss = s |> AsyncSeq.take 1
   let ls = ss |> AsyncSeq.toList
-  ()
+  ()
+
+
+[<Test>]
+let ``AsyncSeqSource.create should create empty sequence`` () =  
+  let src = AsyncSeqSrc.create ()
+  let s = src |> AsyncSeqSrc.toAsyncSeq
+  src |> AsyncSeqSrc.close
+  let expected = AsyncSeq.empty
+  Assert.True(EQ expected s)
+
+
+[<Test>]
+let ``AsyncSeqSource.put should yield`` () =  
+  let item = 1
+  let src = AsyncSeqSrc.create ()  
+  let actual = src |> AsyncSeqSrc.toAsyncSeq  
+  src |> AsyncSeqSrc.put item
+  src |> AsyncSeqSrc.close
+  let expected = AsyncSeq.singleton item
+  Assert.AreEqual (expected, actual)
+
+[<Test>]
+let ``AsyncSeqSource.put should yield after async sequence is created`` () =  
+  let item1 = 1
+  let item2 = 2
+  let src = AsyncSeqSrc.create ()  
+  src |> AsyncSeqSrc.put item1
+  let actual = src |> AsyncSeqSrc.toAsyncSeq  
+  src |> AsyncSeqSrc.put item2
+  src |> AsyncSeqSrc.close
+  let expected = AsyncSeq.ofSeq [item2]
+  Assert.AreEqual (expected, actual)
+
+
+[<Test>]
+let ``AsyncSeq.groupBy should work``() =
+  let ls = List.init 4 id
+  let p i = i % 2
+  let expected = ls |> Seq.groupBy p |> Seq.map (snd >> Seq.toList) |> Seq.toList |> AsyncSeq.ofSeq
+  let actual = ls |> AsyncSeq.ofSeq |> AsyncSeq.groupBy p |> AsyncSeq.mapAsyncParallel (snd >> AsyncSeq.toListAsync)
+  Assert.True(EQ expected actual)