AsyncSeq.bufferByCount / fix toList

Author: eulerfx

src/FSharpx.Async/AsyncSeq.fs

@@ -485,7 +485,9 @@ module AsyncSeq =
 
   /// Creates an async computation which iterates the AsyncSeq and collects the output into a list.
   let toList (input:AsyncSeq<'T>) : Async<'T list> =
-    input |> fold (fun arr a -> a::arr) []
+    input 
+    |> fold (fun arr a -> a::arr) []
+    |> Async.map List.rev
 
   /// Generates an async sequence using the specified generator function.
   let rec unfoldAsync (f:'State -> Async<('T * 'State) option>) (s:'State) : AsyncSeq<'T> = asyncSeq {        
@@ -525,6 +527,31 @@ module AsyncSeq =
     left
 
 
+  /// Buffer items from the async sequence into buffers of a specified size.
+  /// The last buffer returned may be less than the specified buffer size.
+  let rec bufferByCount (bufferSize:int) (s:AsyncSeq<'T>) : AsyncSeq<'T[]> = 
+    if (bufferSize < 1) then invalidArg "bufferSize" "must be positive"
+    async {                  
+      let buffer = ResizeArray<_>()
+      let rec loop s = async {
+        let! step = s
+        match step with
+        | Nil ->
+          if (buffer.Count > 0) then return Cons(buffer.ToArray(),async.Return Nil)
+          else return Nil
+        | Cons(a,tl) ->
+          buffer.Add(a)
+          if buffer.Count = bufferSize then 
+            let buf = buffer.ToArray()
+            buffer.Clear()
+            return Cons(buf, loop tl)
+          else 
+            return! loop tl            
+      }
+      return! loop s
+    }    
+
+
 [<AutoOpen>]
 module AsyncSeqExtensions = 
   /// Builds an asynchronou sequence using the computation builder syntax

tests/FSharpx.Async.Tests/AsyncSeqTests.fs

@@ -78,9 +78,24 @@ let ``unfoldAsync should generate a sequence``() =
 
 
 [<Test>]
-let ``Interleave should interleave two sequences``() =
+let ``interleave should interleave two sequences``() =
   let s1 = AsyncSeq.ofSeq ["a";"b";"c"]
   let s2 = AsyncSeq.ofSeq [1;2;3]
   let merged = AsyncSeq.interleave s1 s2 |> AsyncSeq.toList |> Async.RunSynchronously
   printfn "%A" merged
-  Assert.True([Choice1Of2 "a" ; Choice2Of2 1 ; Choice1Of2 "b" ; Choice2Of2 2 ; Choice1Of2 "c" ; Choice2Of2 3] = merged)
+  Assert.True([Choice1Of2 "a" ; Choice2Of2 1 ; Choice1Of2 "b" ; Choice2Of2 2 ; Choice1Of2 "c" ; Choice2Of2 3] = merged)
+
+
+[<Test>]
+let ``bufferByCount should buffer``() =
+  
+  let s = asyncSeq {
+    yield 1
+    yield 2
+    yield 3
+    yield 4
+  }
+
+  let s' = s |> AsyncSeq.bufferByCount 2 |> AsyncSeq.toList |> Async.RunSynchronously
+
+  Assert.True(([[|1;2|];[|3;4|]] = s'))