cancellable to BlockingSeq

Author: dsyme

src/FSharp.Control.AsyncSeq/Agent.fs

@@ -66,4 +66,5 @@ module internal AsyncExtensions =
             if (Interlocked.CompareExchange(state, 1, 0) = 0) then cnc ex
             else tcs.SetCanceled()
           Async.StartWithContinuations(a, ok, err, cnc)
-          Async.StartWithContinuations(b, ok, err, cnc)
+          Async.StartWithContinuations(b, ok, err, cnc)
+

src/FSharp.Control.AsyncSeq/Async.fs

@@ -350,9 +350,9 @@ module AsyncSeq =
 
   /// Converts observable to an asynchronous sequence. Values that are produced
   /// by the observable while the asynchronous sequence is blocked are discarded
-  /// (this function doesn't guarantee that asynchronou ssequence will return 
+  /// (this function doesn't guarantee that the asynchronous sequence will return 
   /// all values produced by the observable)
-  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
@@ -370,6 +370,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 
+
   /// Converts asynchronous sequence to an IObservable<_>. When the client subscribes
   /// to the observable, a new copy of asynchronous sequence is started and is 
   /// sequentially iterated over (at the maximal possible speed). Disposing of the 
@@ -389,29 +393,36 @@ module AsyncSeq =
   /// Converts asynchronous sequence to a synchronous blocking sequence.
   /// The elements of the asynchronous sequence are consumed lazily.
   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 -> 
+                  match iteratorTask.Result with
+                  | Choice1Of2() -> ()
+                  | Choice2Of2 exn -> raise exn
+              | Some v -> yield v
+      }
 
   /// Create a new asynchronous sequence that caches all elements of the 
   /// sequence specified as the input. When accessing the resulting sequence
   /// multiple times, the input will still be evaluated only once
   let rec cache (input : AsyncSeq<'T>) = 
-    let agent = Agent<AsyncReplyChannel<_>>.Start(fun agent -> async {
+    let agent = MailboxProcessor<AsyncReplyChannel<_>>.Start(fun agent -> async {
       let! (repl:AsyncReplyChannel<AsyncSeqInner<'T>>) = agent.Receive()
       let! next = input
       let res = 

src/FSharp.Control.AsyncSeq/AsyncSeq.fs

@@ -13,13 +13,13 @@ open System.Threading
 /// Wrapper for the standard F# agent (MailboxProcessor) that
 /// supports stopping of the agent's body using the IDisposable 
 /// interface (the type automatically creates a cancellation token)
-type internal AutoCancelAgent<'T> private (mbox:Agent<'T>, cts:CancellationTokenSource) = 
+type internal AutoCancelAgent<'T> private (mbox:MailboxProcessor<'T>, cts:CancellationTokenSource) = 
 
   /// Start a new disposable agent using the specified body function
   /// (the method creates a new cancellation token for the agent)
   static member Start(f) = 
     let cts = new CancellationTokenSource()
-    new AutoCancelAgent<'T>(Agent<'T>.Start(f, cancellationToken = cts.Token), cts)
+    new AutoCancelAgent<'T>(MailboxProcessor<'T>.Start(f, cancellationToken = cts.Token), cts)
 
   /// Returns the number of unprocessed messages in the message queue of the agent.
   member x.CurrentQueueLength = mbox.CurrentQueueLength

src/FSharp.Control.AsyncSeq/AutoCancelAgent.fs

@@ -23,7 +23,7 @@ type internal BlockingQueueAgent<'T>(maxLength) =
     [<VolatileField>]
     let mutable count = 0
 
-    let agent = Agent.Start(fun agent ->
+    let agent = MailboxProcessor.Start(fun agent ->
 
         let queue = new Queue<'T>()
 

src/FSharp.Control.AsyncSeq/BlockingQueueAgent.fs

@@ -59,7 +59,6 @@
   <Import Project="..\..\.paket\paket.targets" />
   <ItemGroup>
     <Compile Include="Utils.fs" />
-    <Compile Include="Agent.fs" />
     <Compile Include="AutoCancelAgent.fs" />
     <Compile Include="BlockingQueueAgent.fs" />
     <Compile Include="Observable.fs" />

src/FSharp.Control.AsyncSeq/FSharp.Control.AsyncSeq.fsproj

@@ -237,7 +237,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>]
@@ -280,4 +285,27 @@ let ``AsyncSeq.skipUntil should not skip with completed signal``() =
 [<Test>]
 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)
+  Assert.True(EQ AsyncSeq.empty actual)
+
+
+[<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)
+