tests, misc docs, pick

Author: eulerfx

src/FSharp.Control.AsyncSeq/AsyncSeq.fs

@@ -706,6 +706,7 @@ module AsyncSeq =
         yield! loop () }
     yield! loop () }
 
+  // TODO
   let mapAsyncParallelBounded (parallelism:int) (f:'a -> Async<'b>) (s:AsyncSeq<'a>) = asyncSeq {
     use mb = BoundedMb.create (parallelism)
     do! s |> iterAsync (fun a -> async {
@@ -794,6 +795,26 @@ module AsyncSeq =
           let! moven = ie.MoveNext()
           b := moven }
 
+  let pickAsync (f:'T -> Async<'U option>) (source:AsyncSeq<'T>) = async { 
+      use ie = source.GetEnumerator() 
+      let! v = ie.MoveNext()
+      let b = ref v
+      let res = ref None
+      while b.Value.IsSome && not res.Value.IsSome do
+          let! fv = f b.Value.Value
+          match fv with 
+          | None -> 
+              let! moven = ie.MoveNext()
+              b := moven
+          | Some _ as r -> 
+              res := r
+      match res.Value with
+      | Some _ -> return res.Value.Value
+      | None -> return raise(KeyNotFoundException()) }
+
+  let pick f (source:AsyncSeq<'T>) =
+    pickAsync (f >> async.Return) source
+
   let tryPickAsync f (source : AsyncSeq<'T>) = async { 
       use ie = source.GetEnumerator() 
       let! v = ie.MoveNext()
@@ -1390,7 +1411,7 @@ module AsyncSeq =
     let close (s:AsyncSeqSrc<'a>) : unit =
       s.tail.tcs.SetResult(None)
 
-    let fail (ex:exn) (s:AsyncSeqSrc<'a>) : unit =
+    let error (ex:exn) (s:AsyncSeqSrc<'a>) : unit =
       s.tail.tcs.SetException(ex)
 
     let rec private toAsyncSeqImpl (s:AsyncSeqSrcNode<'a>) : AsyncSeq<'a> = 
@@ -1462,7 +1483,7 @@ module AsyncSeqSrc =
   let put a s = AsyncSeq.AsyncSeqSrcImpl.put a s
   let close s = AsyncSeq.AsyncSeqSrcImpl.close s
   let toAsyncSeq s = AsyncSeq.AsyncSeqSrcImpl.toAsyncSeq s
-  let fail e s = AsyncSeq.AsyncSeqSrcImpl.fail e s
+  let error e s = AsyncSeq.AsyncSeqSrcImpl.error e s
 
 module Seq = 
 

src/FSharp.Control.AsyncSeq/AsyncSeq.fsi

@@ -19,11 +19,11 @@ type AsyncSeq<'T> = IAsyncEnumerable<'T>
 
 [<RequireQualifiedAccess>]
 module AsyncSeq = 
-    /// Creates an empty asynchronou sequence that immediately ends
+    /// Creates an empty asynchronous sequence that immediately ends.
     [<GeneralizableValueAttribute>]
     val empty<'T> : AsyncSeq<'T>
 
-    /// Creates an asynchronous sequence that generates a single element and then ends
+    /// Creates an asynchronous sequence that generates a single element and then ends.
     val singleton : v:'T -> AsyncSeq<'T>
 
     /// Generates a finite async sequence using the specified asynchronous initialization function.
@@ -32,10 +32,10 @@ module AsyncSeq =
     /// Generates a finite async sequence using the specified initialization function.
     val init : count: int64 -> mapping:(int64 -> 'T) -> AsyncSeq<'T>
 
-    /// Generates an async sequence using the specified asynchronous initialization function.
+    /// Generates an infinite async sequence using the specified asynchronous initialization function.
     val initInfiniteAsync : mapping:(int64 -> Async<'T>) -> AsyncSeq<'T>
 
-    /// Generates an async sequence using the specified initialization function.
+    /// Generates an infinite async sequence using the specified initialization function.
     val initInfinite : mapping:(int64 -> 'T) -> AsyncSeq<'T>
 
     /// Generates an async sequence using the specified asynchronous generator function.
@@ -47,7 +47,7 @@ module AsyncSeq =
     /// Creates an async sequence which repeats the specified value the indicated number of times.
     val replicate : count: int -> v:'T -> AsyncSeq<'T>
 
-    /// Creates an async sequence which repeats the specified value indefinitely.
+    /// Creates an infinite async sequence which repeats the specified value.
     val replicateInfinite : v:'T -> AsyncSeq<'T>
 
     /// Yields all elements of the first asynchronous sequence and then 
@@ -185,9 +185,20 @@ module AsyncSeq =
     /// Asynchronously determine if the sequence contains the given value
     val contains : value:'T -> source:AsyncSeq<'T> -> Async<bool> when 'T : equality
 
-    /// Asynchronously pick a value from a sequence
+    /// Asynchronously pick a value from a sequence based on the specified chooser function.
+    val tryPickAsync : chooser:('T -> Async<'TResult option>) -> source:AsyncSeq<'T> -> Async<'TResult option>
+
+    /// Asynchronously pick a value from a sequence based on the specified chooser function.
     val tryPick : chooser:('T -> 'TResult option) -> source:AsyncSeq<'T> -> Async<'TResult option>
 
+    /// Asynchronously pick a value from a sequence based on the specified chooser function.
+    /// Raises KeyNotFoundException if the chooser function can't find a matching key.
+    val pickAsync : chooser:('T -> Async<'TResult option>) -> source:AsyncSeq<'T> -> Async<'TResult>
+
+    /// Asynchronously pick a value from a sequence based on the specified chooser function.
+    /// Raises KeyNotFoundException if the chooser function can't find a matching key.
+    val pick : chooser:('T -> 'TResult option) -> source:AsyncSeq<'T> -> Async<'TResult>
+
     /// Asynchronously find the first value in a sequence for which the predicate returns true
     val tryFind : predicate:('T -> bool) -> source:AsyncSeq<'T> -> Async<'T option>
 
@@ -420,7 +431,7 @@ module Seq =
     val ofAsyncSeq : source:AsyncSeq<'T> -> seq<'T>
 
 
-/// An async sequence source.
+/// An async sequence source produces async sequences.
 type AsyncSeqSrc<'T>
 
 /// Operations on async sequence sources.
@@ -429,15 +440,16 @@ module AsyncSeqSrc =
   /// Creates a new async sequence source.
   val create : unit -> AsyncSeqSrc<'T>
 
-  /// Puts an item into the async sequence source causing any created async sequences to yield the item.
+  /// Causes any async sequences created before the call to yield the item.
   val put : item:'T -> src:AsyncSeqSrc<'T> -> unit
 
-  /// Closes the async sequence source casuing any created async sequences to terminate.
+  /// Closes the async sequence source casuing any created async sequences to complete.
   val close : src:AsyncSeqSrc<'T> -> unit
 
-  /// Causes async sequence created immediately before the call to raise an exception.
-  val fail : exn:exn -> src:AsyncSeqSrc<'T> -> unit
+  /// Causes async sequence created before the call to raise an exception.
+  val error : exn:exn -> src:AsyncSeqSrc<'T> -> unit
 
   /// Creates an async sequence which yields values as they are put into the source and terminates
   /// when the source is closed. This sequence will yield items starting with the next put.
+  /// Many async sequences can be created from once source.
   val toAsyncSeq : src:AsyncSeqSrc<'T> -> AsyncSeq<'T>

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

@@ -18,6 +18,11 @@ module AsyncOps =
     let unit = async.Return()
     let never = Async.Sleep(-1)
 
+let catch (f:'a -> 'b) : 'a -> Choice<'b, exn> =
+  fun a ->
+    try f a |> Choice1Of2
+    with ex -> ex |> Choice2Of2
+
 /// Determines equality of two async sequences by convering them to lists, ignoring side-effects.
 let EQ (a:AsyncSeq<'a>) (b:AsyncSeq<'a>) = 
   let exp = a |> AsyncSeq.toList 
@@ -49,6 +54,20 @@ type Assert with
       Assert.True((exnEq exp act), message)
     | _ ->
       Assert.Fail(message)
+
+  static member AreEqual (expected:unit -> 'a, actual:unit -> 'a) =
+    let expected = (catch expected) ()
+    let actual = (catch actual) ()
+    let message = sprintf "expected=%A actual=%A" expected actual
+    match expected,actual with
+    | Choice1Of2 exp, Choice1Of2 act ->
+      Assert.True((exp = act), message)
+    | Choice2Of2 exp, Choice2Of2 act ->
+      ()
+    | _ ->
+      Assert.Fail(message)
+
+    
 
 
 
@@ -117,6 +136,16 @@ let ``AsyncSeq.tryPick works``() =
           let expected = ls |> Seq.tryPick (fun x -> if x = j then Some (string (x+1)) else None)
           Assert.True((expected = actual))
 
+[<Test>]
+let ``AsyncSeq.pick works``() =  
+  for i in 0 .. 10 do 
+      let ls = [ 1 .. i ]
+      for j in [0;i;i+1] do
+          let chooser x = if x = j then Some (string (x+1)) else None
+          let actual () = AsyncSeq.ofSeq ls |> AsyncSeq.pick chooser |> Async.RunSynchronously
+          let expected () = ls |> Seq.pick chooser
+          Assert.AreEqual(actual, expected)
+
 [<Test>]
 let ``AsyncSeq.tryFind works``() =  
   for i in 0 .. 10 do 
@@ -1093,16 +1122,17 @@ let ``AsyncSeq.take should work``() =
 
 [<Test>]
 let ``AsyncSeq.mapParallelAsync should maintain order`` () =
-  let ls = List.init 500 id
-  let expected = 
-    ls 
-    |> AsyncSeq.ofSeq
-    |> AsyncSeq.mapAsync (async.Return)
-  let actual = 
-    ls 
-    |> AsyncSeq.ofSeq 
-    |> AsyncSeq.mapAsyncParallel (async.Return)
-  Assert.AreEqual(expected, actual)
+  for i in 0..100 do
+    let ls = List.init i id
+    let expected = 
+      ls 
+      |> AsyncSeq.ofSeq
+      |> AsyncSeq.mapAsync (async.Return)
+    let actual = 
+      ls 
+      |> AsyncSeq.ofSeq 
+      |> AsyncSeq.mapAsyncParallel (async.Return)
+    Assert.AreEqual(expected, actual)
 
 //[<Test>]
 let ``AsyncSeq.mapParallelAsync should be parallel`` () =
@@ -1113,7 +1143,7 @@ let ``AsyncSeq.mapParallelAsync should be parallel`` () =
     s |> AsyncSeq.map id
   let actual = 
     s
-    |> AsyncSeq.mapAsyncParallel (fun i -> async { barrier.SignalAndWait () ; return i })
+    |> AsyncSeq.mapAsyncParallel (fun i -> async { barrier.SignalAndWait () ; return i }) // can deadlock
   Assert.AreEqual(expected, actual, timeout=200)
 
 //[<Test>]
@@ -1164,27 +1194,29 @@ let ``AsyncSeqSrc.fail should throw`` () =
   let item = 1
   let src = AsyncSeqSrc.create ()    
   let actual = src |> AsyncSeqSrc.toAsyncSeq
-  src |> AsyncSeqSrc.fail (exn("test"))  
+  src |> AsyncSeqSrc.error (exn("test"))  
   let expected = asyncSeq { raise (exn("test")) }
   Assert.AreEqual (expected, actual)
 
 
 [<Test>]
 let ``AsyncSeq.groupBy should work``() =
-  let ls = List.init 100 id
-  let p i = i % 3
-  let expected = 
-    ls 
-    |> Seq.groupBy p 
-    |> Seq.map (snd >> Seq.toList) 
-    |> Seq.toList 
-    |> AsyncSeq.ofSeq
-  let actual = 
-    ls 
-    |> AsyncSeq.ofSeq 
-    |> AsyncSeq.groupBy p 
-    |> AsyncSeq.mapAsyncParallel (snd >> AsyncSeq.toListAsync)
-  Assert.AreEqual(expected, actual)
+  for i in 0..100 do
+    for j in 1..3 do
+      let ls = List.init i id
+      let p x = x % j
+      let expected = 
+        ls 
+        |> Seq.groupBy p 
+        |> Seq.map (snd >> Seq.toList) 
+        |> Seq.toList 
+        |> AsyncSeq.ofSeq
+      let actual = 
+        ls 
+        |> AsyncSeq.ofSeq 
+        |> AsyncSeq.groupBy p 
+        |> AsyncSeq.mapAsyncParallel (snd >> AsyncSeq.toListAsync)
+      Assert.AreEqual(expected, actual)
 
 [<Test>]
 let ``AsyncSeq.groupBy should propagate exception and terminate all groups``() =