Merge pull request #10 from fsprojects/fix-4

Fix #9 - toBlockingSeq doesn't cancel the task

Author: dsyme

src/FSharp.Control.AsyncSeq/AsyncSeq.fs

@@ -351,7 +351,7 @@ module AsyncSeq =
             repls.Dequeue().Reply(buffer.Dequeue())  })
 
 
-  let ofObservable (input : System.IObservable<_>) = 
+  let ofObservableDiscarding (input : System.IObservable<_>) = 
     ofObservableUsingAgent input (fun mbox -> async {
       while true do 
         // Allow timeout (when the observable ends, caller will
@@ -369,6 +369,10 @@ module AsyncSeq =
             | Put v -> repl.Reply(v)
             | _ -> failwith "Unexpected Get" })
 
+  [<System.Obsolete("Use AsyncSeq.ofObservableDiscarding. This function doesn't guarantee that the asynchronous sequence will return all values produced by the observable")>]
+  let ofObservable (input : System.IObservable<_>) = 
+      ofObservableDiscarding input 
+
   let toObservable (aseq:AsyncSeq<_>) =
     let start (obs:IObserver<_>) =
       async {
@@ -382,23 +386,31 @@ module AsyncSeq =
         member x.Subscribe(obs) = start obs }
 
   let toBlockingSeq (input : AsyncSeq<'T>) = 
-    // Write all elements to a blocking buffer and then add None to denote end
-    let buf = new BlockingQueueAgent<_>(1)
-    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 -> iterator.Result
-      | Some v -> 
-          yield v
-          yield! loop() }
-    loop ()
+      seq { 
+          // Write all elements to a blocking buffer and then add None to denote end
+          let buf = new BlockingQueueAgent<_>(1)
+          
+          use cts = new System.Threading.CancellationTokenSource()
+          use _cancel = { new IDisposable with member __.Dispose() = cts.Cancel() }
+          let iteratorTask = 
+              async { 
+                  let! res = iterAsync (Some >> buf.AsyncAdd) input |> Async.Catch
+                  do! buf.AsyncAdd(None)
+                  return res
+              }
+              |> fun p -> Async.StartAsTask(p, cancellationToken = cts.Token)
+          
+          // Read elements from the blocking buffer & return a sequences
+          let fin = ref false
+          while not fin.Value do
+              match buf.Get() with
+              | None -> 
+                  fin := true
+                  match iteratorTask.Result with
+                  | Choice1Of2() -> ()
+                  | Choice2Of2 exn -> raise exn
+              | Some v -> yield v
+      }
 
   let rec cache (input : AsyncSeq<'T>) = 
     let agent = MailboxProcessor<AsyncReplyChannel<_>>.Start(fun agent -> async {

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

@@ -250,7 +250,12 @@ let ``AsyncSeq.toBlockingSeq does not hung forever and rethrows exception``() =
       yield 1
       failwith "error"
   }
-  Assert.Throws<AggregateException>(fun _ -> s |> AsyncSeq.toBlockingSeq |> Seq.toList |> ignore) |> ignore
+  Assert.Throws<Exception>(fun _ -> s |> AsyncSeq.toBlockingSeq |> Seq.toList |> ignore) |> ignore
+  try 
+      let _ = s |> AsyncSeq.toBlockingSeq |> Seq.toList 
+      ()
+  with e -> 
+      Assert.AreEqual(e.Message, "error") 
 
 
 [<Test>]
@@ -295,6 +300,63 @@ let ``AsyncSeq.skipUntil should skip everything with never signal``() =
   let actual = [1;2;3;4] |> AsyncSeq.ofSeq |> AsyncSeq.skipUntil AsyncOps.never
   Assert.True(EQ AsyncSeq.empty actual)
 
+[<Test>]
+let ``AsyncSeq.toBlockingSeq should work length 1``() =
+  let s = asyncSeq { yield 1 } |> AsyncSeq.toBlockingSeq  |> Seq.toList
+  Assert.True((s = [1]))
+
+[<Test>]
+let ``AsyncSeq.toBlockingSeq should work length 0``() =
+  let s = asyncSeq { () } |> AsyncSeq.toBlockingSeq  |> Seq.toList
+  Assert.True((s = []))
+
+[<Test>]
+let ``AsyncSeq.toBlockingSeq should work length 2 with sleep``() =
+  let s = asyncSeq { yield 1 
+                     do! Async.Sleep 10
+                     yield 2  } |> AsyncSeq.toBlockingSeq  |> Seq.toList
+  Assert.True((s = [1; 2]))
+
+[<Test>]
+let ``AsyncSeq.toBlockingSeq should work length 1 with fail``() =
+  let s = 
+      asyncSeq { yield 1 
+                 failwith "fail"  } 
+      |> AsyncSeq.toBlockingSeq  
+      |> Seq.truncate 1 
+      |> Seq.toList
+  Assert.True((s = [1]))
+
+[<Test>]
+let ``AsyncSeq.toBlockingSeq should work length 0 with fail``() =
+  let s = 
+      asyncSeq { failwith "fail"  } 
+      |> AsyncSeq.toBlockingSeq  
+      |> Seq.truncate 0 
+      |> Seq.toList
+  Assert.True((s = []))
+
+[<Test>]
+let ``AsyncSeq.toBlockingSeq should be cancellable``() =
+  let cancelCount = ref 0 
+  let aseq = 
+      asyncSeq { 
+          use! a = Async.OnCancel(fun x -> incr cancelCount)
+          while true do 
+              yield 1 
+              do! Async.Sleep 10
+    }
+    
+  let asSeq = aseq |> AsyncSeq.toBlockingSeq
+  let enum = asSeq.GetEnumerator()
+  Assert.AreEqual(cancelCount.Value, 0)
+  let canMoveNext = enum.MoveNext()
+  Assert.AreEqual(canMoveNext, true)
+  Assert.AreEqual(cancelCount.Value, 0)
+  enum.Dispose()
+  System.Threading.Thread.Sleep(1000) // wait for task cancellation to be effective
+  Assert.AreEqual(cancelCount.Value, 1)
+
 [<Test>]
 let ``AsyncSeq.while should allow do at end``() =  
   let s1 = asyncSeq {