Add lwt aware stream module for IO

rock/src/io_stream.ml

@@ -0,0 +1,57 @@
+open Lwt.Syntax
+
+module Input = struct
+  type 'a t = unit -> 'a option Lwt.t
+
+  let create f = f
+  let read f = f ()
+
+  let rec iter f t =
+    let* res = read t in
+    match res with
+    | None -> Lwt.return_unit
+    | Some item ->
+      let* () = f item in
+      iter f t
+  ;;
+
+  let singleton item =
+    let finished = ref false in
+    fun () ->
+      if !finished
+      then Lwt.return_none
+      else (
+        finished := true;
+        Lwt.return (Some item))
+  ;;
+end
+
+module Output = struct
+  type 'a t = 'a option -> unit Lwt.t
+
+  let create f =
+    let closed = ref false in
+    fun v ->
+      if !closed
+      then Lwt.return_unit
+      else (
+        match v with
+        | None ->
+          closed := true;
+          f None
+        | v -> f v)
+  ;;
+
+  let write i f = f i
+end
+
+let transfer input output =
+  let rec loop () =
+    let* item = Input.read input in
+    let* () = Output.write item output in
+    match item with
+    | None -> Lwt.return_unit
+    | Some _ -> loop ()
+  in
+  loop ()
+;;

rock/src/io_stream.mli

@@ -0,0 +1,17 @@
+module Input : sig
+  type 'a t
+
+  val create : (unit -> 'a option Lwt.t) -> 'a t
+  val singleton : 'a -> 'a t
+  val read : 'a t -> 'a option Lwt.t
+  val iter : ('a -> unit Lwt.t) -> 'a t -> unit Lwt.t
+end
+
+module Output : sig
+  type 'a t
+
+  val create : ('a option -> unit Lwt.t) -> 'a t
+  val write : 'a option -> 'a t -> unit Lwt.t
+end
+
+val transfer : 'a Input.t -> 'a Output.t -> unit Lwt.t

rock/src/rock.ml

@@ -1,5 +1,6 @@
 module App = App
 module Context = Context
+module Io_stream = Io_stream
 module Body = Body
 module Request = Request
 module Response = Response

rock/src/rock.mli

@@ -6,6 +6,7 @@ module App = App
 module Context = Context
 module Request = Request
 module Response = Response
+module Io_stream = Io_stream
 module Body = Body
 module Service = Service
 module Filter = Filter