Add stream function to get answers one-by one with continuation

src/core/Stream.ml

@@ -197,3 +197,14 @@ let rec retrieve ?(n=(-1)) s =
   | Some (x, s)  -> let xs, s = retrieve ~n:(n-1) s in x::xs, s
 
 let take ?n s = fst @@ retrieve ?n s
+
+let iteri_k n stream fin f =
+  let rec helper i n stream ~f =
+    if i > n then fin (i - 1)
+    else
+      f i
+        (fun () -> msplit stream)
+        (fun tl -> helper (1 + i) n ~f tl)
+  in
+  helper 0 n ~f stream
+

src/core/Stream.mli

@@ -92,6 +92,17 @@ val tl : 'a t -> 'a t
 
 val msplit: 'a t -> ('a * 'a t) option
 
+(** [iteri_k n stream fin k] iterates over [stream] trying to get first [n] answers.
+
+  Answers are extracted one by one using !{msplit} and are possed to continuation function.
+  If all answers are received, then [fin (n-1)] is executed.
+  The continuation function receives delay stream extract and a continuation to continue
+  processin the tail of the stream.
+*)
+val iteri_k: int -> 'a t -> (int -> 'b) ->
+  (int -> (unit -> ('a * 'a t) option) -> ('a t -> 'b) -> 'b) ->
+  'b
+
 IFDEF STATS THEN
 (* Gets a counter *)
 val unwrap_suspended_counter : unit -> int