Change interface, update some stuff, new example...

1 files changed, 75 insertions, 0 deletions

diff --git a/src/kahn_pipe.ml b/src/kahn_pipe.ml
new file mode 100644
index 0000000..f0bec97
--- /dev/null
+++ b/src/kahn_pipe.ml
@@ -0,0 +1,75 @@
+open Kahn
+
+module Pipe: S = struct
+	type 'a process = unit -> 'a
+
+	type 'a in_port = in_channel
+	type 'a out_port = out_channel
+
+	let new_channel =
+		fun () ->
+			let i, o = Unix.pipe () in
+			Unix.in_channel_of_descr i, Unix.out_channel_of_descr o
+
+	let get c =
+		fun () -> Marshal.from_channel c
+
+	let put x c =
+		fun () ->
+      Marshal.to_channel c x [];
+      flush c
+	
+	let try_get block prt_list =
+		let fds = List.map fst prt_list in
+		let fds = List.map Unix.descr_of_in_channel fds in
+		let ok_fds, _, _ = Unix.select fds [] []
+			(if block then -1.0 else 0.0) 
+		in
+		match ok_fds with
+		| [] -> None
+		| fd::x ->
+			let chan, f =
+				List.find
+					(fun (s, _) -> Unix.descr_of_in_channel s = fd)
+					prt_list
+			in
+				Some(f (Marshal.from_channel chan))
+	
+	let select prt_list =
+		fun () ->
+			match try_get true prt_list with
+			| Some x -> x
+			| None -> assert false
+
+	let select_default prt_list def =
+		fun () ->
+			match try_get false prt_list with
+			| Some x -> x
+			| None -> def ()
+
+	let return v =
+		fun () -> v
+
+	let bind e f =
+		fun () -> f (e ()) ()
+	let bind_io = bind
+
+	let run p =
+    p()
+
+	let doco l =
+		fun () ->
+      let children =
+        List.map
+          (fun p ->
+            match Unix.fork () with
+            | 0 ->
+              run p;
+              exit 0
+            | i -> i)
+          l
+      in
+        List.iter
+          (fun x -> try ignore(Unix.waitpid [] x) with _ -> ())
+          children
+end