bufferByCountAndTime

Author: eulerfx

src/FSharp.Control.AsyncSeq/AsyncSeq.fs

@@ -986,6 +986,44 @@ module AsyncSeq =
       if (buffer.Count > 0) then 
           yield buffer.ToArray() }
 
+  let bufferByCountAndTime (bufferSize:int) (timeoutMs:int) (source:AsyncSeq<'T>) : AsyncSeq<'T[]> = 
+    if (bufferSize < 1) then invalidArg "bufferSize" "must be positive"
+    if (timeoutMs < 1) then invalidArg "timeoutMs" "must be positive"
+    asyncSeq {
+      let buffer = new ResizeArray<_>()
+      use ie = source.GetEnumerator()
+      let rec loop rem rt = asyncSeq {
+        let! move = 
+          match rem with
+          | Some rem -> async.Return rem
+          | None -> Async.StartChildAsTask(ie.MoveNext())
+        let t = DateTime.Now
+        let! time = Async.StartChildAsTask(Async.Sleep (max 0 rt))
+        let! moveOr = Async.chooseTasks move time
+        let delta = int (DateTime.Now - t).TotalMilliseconds      
+        match moveOr with
+        | Choice1Of2 (None, _) -> 
+          if buffer.Count > 0 then
+            yield buffer.ToArray()
+        | Choice1Of2 (Some v, _) ->
+          buffer.Add v
+          if buffer.Count = bufferSize then
+            yield buffer.ToArray()
+            buffer.Clear()
+            yield! loop None timeoutMs
+          else
+            yield! loop None (rt - delta)
+        | Choice2Of2 (_, rest) ->
+          if buffer.Count > 0 then
+            yield buffer.ToArray()
+            buffer.Clear()
+            yield! loop (Some rest) timeoutMs
+          else
+            yield! loop (Some rest) timeoutMs
+      }
+      yield! loop None timeoutMs
+    }
+
   let mergeChoice (source1:AsyncSeq<'T1>) (source2:AsyncSeq<'T2>) : AsyncSeq<Choice<'T1,'T2>> = asyncSeq {
       use ie1 = source1.GetEnumerator() 
       use ie2 = source2.GetEnumerator() 

src/FSharp.Control.AsyncSeq/AsyncSeq.fsi

@@ -106,11 +106,11 @@ module AsyncSeq =
         member Zero : unit -> AsyncSeq<'T>
 
 
-  /// Creates an asynchronous sequence that iterates over the given input sequence.
-  /// For every input element, it calls the the specified function and iterates
-  /// over all elements generated by that asynchronous sequence.
-  /// This is the 'bind' operation of the computation expression (exposed using
-  /// the 'for' keyword in asyncSeq computation).
+    /// Creates an asynchronous sequence that iterates over the given input sequence.
+    /// For every input element, it calls the the specified function and iterates
+    /// over all elements generated by that asynchronous sequence.
+    /// This is the 'bind' operation of the computation expression (exposed using
+    /// the 'for' keyword in asyncSeq computation).
     val collect : mapping:('T -> AsyncSeq<'TResult>) -> source:AsyncSeq<'T> -> AsyncSeq<'TResult>
 
     /// Builds a new asynchronous sequence whose elements are generated by 
@@ -334,6 +334,9 @@ module AsyncSeq =
     /// The last buffer returned may be less than the specified buffer size.
     val bufferByCount : bufferSize:int -> source:AsyncSeq<'T> -> AsyncSeq<'T []>
 
+    /// Buffer items from the async sequence until a specified buffer size is reached or a specified amount of time is elapsed.
+    val bufferByCountAndTime : bufferSize:int -> timeoutMs:int -> source:AsyncSeq<'T> -> AsyncSeq<'T []>
+
     /// Merges two async sequences into an async sequence non-deterministically.
     val mergeChoice: source1:AsyncSeq<'T1> -> source2:AsyncSeq<'T2> -> AsyncSeq<Choice<'T1,'T2>>
 

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

@@ -205,6 +205,27 @@ let ``AsyncSeq.bufferByCount empty``() =
   let s' = s |> AsyncSeq.bufferByCount 2 |> AsyncSeq.toList 
   Assert.True(([] = s'))
 
+
+[<Test>]
+let ``AsyncSeq.bufferByTimeAndCount``() =      
+  let s = asyncSeq {
+    yield 1
+    yield 2
+    yield 3
+    do! Async.Sleep 250
+    yield 4
+    yield 5
+  }
+  let actual = AsyncSeq.bufferByCountAndTime 2 50 s |> AsyncSeq.toList
+  Assert.True((actual = [ [|1;2|] ; [|3|] ; [|4;5|] ]))
+
+[<Test>]
+let ``AsyncSeq.bufferByTimeAndCount empty``() =      
+  let s = AsyncSeq.empty<int>
+  let actual = AsyncSeq.bufferByCountAndTime 2 10 s |> AsyncSeq.toList
+  Assert.True((actual = []))
+
+
 [<Test>]
 let ``try finally works no exception``() =
   let x = ref 0