merge upstream

Author: eulerfx

src/FSharpx.Async/AssemblyInfo.fs

@@ -4,9 +4,9 @@ open System.Reflection
 [<assembly: AssemblyTitleAttribute("FSharpx.Async")>]
 [<assembly: AssemblyProductAttribute("FSharpx.Async")>]
 [<assembly: AssemblyDescriptionAttribute("Async extensions for F#")>]
-[<assembly: AssemblyVersionAttribute("1.11.0")>]
-[<assembly: AssemblyFileVersionAttribute("1.11.0")>]
+[<assembly: AssemblyVersionAttribute("1.11.1")>]
+[<assembly: AssemblyFileVersionAttribute("1.11.1")>]
 do ()
 
 module internal AssemblyVersionInformation =
-    let [<Literal>] Version = "1.11.0"
+    let [<Literal>] Version = "1.11.1"

src/FSharpx.Async/AsyncSeq.fs

@@ -390,14 +390,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() }
@@ -674,4 +677,4 @@ module Seq =
   /// Converts asynchronous sequence to a synchronous blocking sequence.
   /// The elements of the asynchronous sequence are consumed lazily.
   let ofAsyncSeq (input : AsyncSeq<'T>) =
-    AsyncSeq.toBlockingSeq input
+    AsyncSeq.toBlockingSeq input

tests/FSharpx.Async.Tests/AsyncSeqTests.fs

@@ -2,6 +2,7 @@
 
 open NUnit.Framework
 open FSharpx.Control
+open System
 
 /// Determines equality of two async sequences by convering them to lists, ignoring side-effects.
 let EQ (a:AsyncSeq<'a>) (b:AsyncSeq<'a>) = 
@@ -236,4 +237,12 @@ let ``AsyncSeq.traverseChoiceAsync``() =
   | Choice2Of2 e -> 
     Assert.AreEqual("oh no", e)
     Assert.True(([1;2] = (seen |> List.ofSeq)))
-  
+  
+
+[<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