Added Stream.bufferTime see #132

Libs/Hopac/Stream.fs

@@ -844,6 +844,31 @@ module Stream =
     if n < 1 then failwith "buffer: n < 1" else
     buffer' 0 (Array.zeroCreate n) xs |> memo
 
+  let rec bufferTime' n (b: ResizeArray<_>) timeOut xs clock =
+        Alt.tryIn xs
+          <| function Cons (x, xs) ->
+                      b.Add x
+                      if b.Count < n
+                      then bufferTime' n b timeOut xs clock
+                      else consj b ^ bufferTime n timeOut xs
+                    | Nil -> consj b nil
+          <| fun e -> consj b ^ error e
+    <|> clock ^=> fun _ ->
+          consj b ^ bufferTime n timeOut xs
+     :> Job<_>
+  and bufferTime n timeOut xs =
+    if n < 1 then failwith "bufferTime: n < 1" else
+    Job.tryIn xs
+     <| function Cons (x, xs) ->
+                 let b = ResizeArray<_> (1)
+                 b.Add x
+                 if n <> 1
+                 then bufferTime' n b timeOut xs ^ memo timeOut
+                 else consj b ^ bufferTime n timeOut xs
+               | Nil -> nilj
+     <| error
+    |> memo
+
   let values (xs: Stream<'x>) : Alt<'x> =
     let vs = Ch ()
     let (inc, dec) =

Libs/Hopac/Stream.fsi

@@ -817,6 +817,20 @@ module Stream =
   /// at most given number of elements.
   val buffer: int -> Stream<'x> -> Stream<array<'x>>
 
+  /// Converts a stream of elements into a stream of non-overlapping buffers of
+  /// at most given number of elements obtained within the given time limit.  To
+  /// avoid busy-wait polling, the time limit is started when the first element
+  /// for a new buffer is obtained.
+#if DOC
+  ///
+  /// Here is an example diagram with `bufferTime 3`:
+  ///
+  ///>   input: 1        234 56
+  ///> timeout: +-----x  +-! +-----x
+  ///>  output:      [1] [234]   [56]
+#endif
+  val bufferTime: int -> timeout: Job<_> -> Stream<'x> -> Stream<ResizeArray<'x>>
+
   //# Accumulating state
 
   /// Returns a stream whose elements are computed using the given job and