AsyncSeq.toBlockingSeq does not hung forever if an exception is thrown and reraise it outside

Author: Vasily Kirichenko

src/FSharpx.Async/AsyncSeq.fs

@@ -378,14 +378,17 @@ module AsyncSeq =
   let toBlockingSeq (input : AsyncSeq<'T>) = 
     // Write all elements to a blocking buffer and then add None to denote end
     let buf = new BlockingQueueAgent<_>(1)
-    async {
-      do! iterAsync (Some >> buf.AsyncAdd) input
-      do! buf.AsyncAdd(None) } |> Async.Start
+    let iterator =
+        async {
+            let! res = iterAsync (Some >> buf.AsyncAdd) input |> Async.Catch
+            do! buf.AsyncAdd(None)
+            return match res with Choice2Of2 e -> raise e | _ -> () 
+        } |> Async.StartAsTask
 
     // Read elements from the blocking buffer & return a sequences
     let rec loop () = seq {
       match buf.Get() with
-      | None -> ()
+      | None -> iterator.Result
       | Some v -> 
           yield v
           yield! loop() }

tests/FSharpx.Async.Tests/AsyncSeqTests.fs

@@ -2,6 +2,7 @@
 
 open NUnit.Framework
 open FSharpx.Control
+open System
 
 [<Test>]
 let ``skipping should return all elements after the first non-match``() =
@@ -98,4 +99,12 @@ let ``AsyncSeq.bufferByCount``() =
 
   let s' = s |> AsyncSeq.bufferByCount 2 |> AsyncSeq.toList |> Async.RunSynchronously
 
-  Assert.True(([[|1;2|];[|3;4|]] = s'))
+  Assert.True(([[|1;2|];[|3;4|]] = s'))
+
+[<Test>]
+let ``AsyncSeq.toBlockingSeq does not hung forever and rethrows exception``() =
+  let s = asyncSeq {
+      yield 1
+      failwith "error"
+  }
+  Assert.Throws<AggregateException>(fun _ -> s |> AsyncSeq.toBlockingSeq |> Seq.toList |> ignore) |> ignore