Add lwt aware stream module for IO

dune-project

@@ -81,3 +81,13 @@
    (= :version))
   alcotest
   alcotest-lwt))
+
+(package
+  (name io_stream)
+  (synopsis "Streaming I/O")
+  (depends
+    (ocaml
+      (>= 4.08))
+    (dune
+      (>= 1.11))
+    lwt))

io_stream.opam

@@ -0,0 +1,29 @@
+# This file is generated by dune, edit dune-project instead
+opam-version: "2.0"
+synopsis: "Streaming I/O"
+maintainer: ["Rudi Grinberg <[email protected]>"]
+authors: ["Rudi Grinberg" "Anurag Soni" "Thibaut Mattio"]
+license: "MIT"
+homepage: "https://github.com/rgrinberg/opium"
+doc: "https://rgrinberg.github.io/opium/"
+bug-reports: "https://github.com/rgrinberg/opium/issues"
+depends: [
+  "ocaml" {>= "4.08"}
+  "dune" {>= "1.11"}
+  "lwt"
+]
+build: [
+  ["dune" "subst"] {pinned}
+  [
+    "dune"
+    "build"
+    "-p"
+    name
+    "-j"
+    jobs
+    "@install"
+    "@runtest" {with-test}
+    "@doc" {with-doc}
+  ]
+]
+dev-repo: "git+https://github.com/rgrinberg/opium.git"

io_stream/dune

@@ -0,0 +1,4 @@
+(library
+ (name io_stream)
+ (public_name io_stream)
+ (libraries lwt))

io_stream/io_stream.ml

@@ -0,0 +1,57 @@
+open Lwt.Syntax
+
+module In = 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 of_list xs =
+    let xs = ref xs in
+    fun () ->
+      match !xs with
+      | [] -> Lwt.return_none
+      | x :: xs' ->
+        xs := xs';
+        Lwt.return_some x
+  ;;
+end
+
+module Out = 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 connect input output =
+  let rec loop () =
+    let* item = In.read input in
+    let* () = Out.write item output in
+    match item with
+    | None -> Lwt.return_unit
+    | Some _ -> loop ()
+  in
+  loop ()
+;;

io_stream/io_stream.mli

@@ -0,0 +1,17 @@
+module In : sig
+  type 'a t
+
+  val create : (unit -> 'a option Lwt.t) -> 'a t
+  val of_list : 'a list -> 'a t
+  val read : 'a t -> 'a option Lwt.t
+  val iter : ('a -> unit Lwt.t) -> 'a t -> unit Lwt.t
+end
+
+module Out : sig
+  type 'a t
+
+  val create : ('a option -> unit Lwt.t) -> 'a t
+  val write : 'a option -> 'a t -> unit Lwt.t
+end
+
+val connect : 'a In.t -> 'a Out.t -> unit Lwt.t