simplify ofObservableBuffered

Author: dsyme

src/FSharp.Control.AsyncSeq/AsyncSeq.fs

@@ -335,20 +335,21 @@ module AsyncSeq =
       yield! loop() }
 
   let ofObservableBuffered (input : System.IObservable<_>) = 
-    ofObservableUsingAgent input (fun mbox -> async {
-        let buffer = new System.Collections.Generic.Queue<_>()
-        let repls = new System.Collections.Generic.Queue<_>()
-        while true do
-          // Receive next message (when observable ends, caller will
-          // cancel the agent, so we need timeout to allow cancleation)
-          let! msg = mbox.TryReceive(200)
-          match msg with 
-          | Some(Put(v)) -> buffer.Enqueue(v)
-          | Some(Get(repl)) -> repls.Enqueue(repl)
-          | _ -> () 
-          // Process matching calls from buffers
-          while buffer.Count > 0 && repls.Count > 0 do
-            repls.Dequeue().Reply(buffer.Dequeue())  })
+    asyncSeq {  
+      let cts = new CancellationTokenSource()
+      try 
+        let agent = MailboxProcessor<_>.Start((fun inbox -> async.Return() ), cancellationToken = cts.Token)
+        use d = input |> Observable.asUpdates |> Observable.subscribe agent.Post
+        let fin = ref false
+        while not fin.Value do 
+          let! msg = agent.Receive()
+          match msg with
+          | Observable.ObservableUpdate.Error e -> raise e
+          | Observable.Completed -> fin := true
+          | Observable.Next v -> yield v 
+      finally 
+         // Cancel on early exit 
+         cts.Cancel() }
 
 
   let ofObservableDiscarding (input : System.IObservable<_>) = 
@@ -369,7 +370,7 @@ 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")>]
+  [<System.Obsolete("Use AsyncSeq.ofObservableDiscarding or AsyncSeq.ofObservableBuffered. This function doesn't guarantee that the asynchronous sequence will return all values produced by the observable")>]
   let ofObservable (input : System.IObservable<_>) = 
       ofObservableDiscarding input 
 

src/FSharp.Control.AsyncSeq/AsyncSeq.fsi

@@ -185,6 +185,9 @@ module AsyncSeq =
     /// by the observable while the asynchronous sequence is blocked are discarded
     /// (this function doesn't guarantee that asynchronou ssequence will return 
     /// all values produced by the observable)
+    val ofObservableDiscarding : input:System.IObservable<'T> -> AsyncSeq<'T>
+
+    [<System.Obsolete("Use AsyncSeq.ofObservableDiscarding or AsyncSeq.ofObservableBuffered. This function doesn't guarantee that the asynchronous sequence will return all values produced by the observable")>]
     val ofObservable : input:System.IObservable<'T> -> AsyncSeq<'T>
 
     /// Converts asynchronous sequence to an IObservable<_>. When the client subscribes

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

@@ -365,3 +365,37 @@ let ``AsyncSeq.while should allow do at end``() =
         do! Async.Sleep 10
   }
   Assert.True(true)
+
+let observe vs err = 
+    { new IObservable<'U> with 
+            member x.Subscribe(observer) = 
+                for v in vs do 
+                   observer.OnNext v
+                if err then 
+                   observer.OnError (Failure "fail")
+                observer.OnCompleted()
+                { new IDisposable with member __.Dispose() = () }  }
+    
+[<Test>]
+let ``AsyncSeq.ofObservable should work (empty)``() =  
+  Assert.True(observe [] false |> AsyncSeq.ofObservableBuffered |> AsyncSeq.toList |> Async.RunSynchronously = [])
+
+[<Test>]
+let ``AsyncSeq.ofObservable should work (sinngleton)``() =  
+  Assert.True(observe [1] false |> AsyncSeq.ofObservableBuffered |> AsyncSeq.toList |> Async.RunSynchronously = [1])
+
+[<Test>]
+let ``AsyncSeq.ofObservable should work (ten)``() =  
+  Assert.True(observe [1..10] false |> AsyncSeq.ofObservableBuffered |> AsyncSeq.toList |> Async.RunSynchronously = [1..10])
+
+[<Test>]
+let ``AsyncSeq.ofObservable should work (empty, fail)``() =  
+  Assert.True(try (observe [] true |> AsyncSeq.ofObservableBuffered |> AsyncSeq.toList |> Async.RunSynchronously |> ignore); false with _ -> true)
+
+[<Test>]
+let ``AsyncSeq.ofObservable should work (one, fail)``() =  
+  Assert.True(try (observe [1] true |> AsyncSeq.ofObservableBuffered |> AsyncSeq.toList |> Async.RunSynchronously |> ignore); false with _ -> true)
+
+[<Test>]
+let ``AsyncSeq.ofObservable should work (one, take)``() =  
+  Assert.True(observe [1] false |> AsyncSeq.ofObservableBuffered |> AsyncSeq.take 1 |> AsyncSeq.toList |> Async.RunSynchronously = [1])