pipes

libs/uv/uv.ml

@@ -245,10 +245,14 @@ external timer_stop : t_timer -> unit_cb -> unit uv_result = "w_timer_stop"
 type process_cb = (int * int) uv_result -> unit
 
 type process_io =
-	| UvIoPipe of bool * bool
+	| UvIoPipe of bool * bool * t_pipe
 	| UvIoIgnore
 	| UvIoInherit
 
 external spawn : t_loop -> process_cb -> string -> string array -> string array -> string -> int -> process_io array -> int -> int -> t_process uv_result = "w_spawn_bytecode" "w_spawn"
 external process_kill : t_process -> int -> unit uv_result = "w_process_kill"
 external process_get_pid : t_process -> int = "w_process_get_pid"
+
+(* ------------- PIPES ---------------------------------------------- *)
+
+external pipe_init : t_loop -> bool -> t_pipe uv_result = "w_pipe_init"

libs/uv/uv_stubs.c

@@ -41,6 +41,7 @@
 #define GetAddrInfo_val(v) UV_UNWRAP(v, uv_getaddrinfo_t)
 #define Handle_val(v) UV_UNWRAP(v, uv_handle_t)
 #define Loop_val(v) UV_UNWRAP(v, uv_loop_t)
+#define Pipe_val(v) UV_UNWRAP(v, uv_pipe_t)
 #define Process_val(v) UV_UNWRAP(v, uv_process_t)
 #define Shutdown_val(v) UV_UNWRAP(v, uv_shutdown_t)
 #define Stream_val(v) UV_UNWRAP(v, uv_stream_t)
@@ -402,6 +403,10 @@ typedef struct {
 			value cb_exit;
 			value unused1;
 		} process;
+		struct {
+			value unused1;
+			value unused2;
+		} pipe;
 	} u;
 } uv_w_handle_t;
 
@@ -462,6 +467,15 @@ static uv_w_handle_t *alloc_data_process(value cb_exit) {
 	return data;
 }
 
+static uv_w_handle_t *alloc_data_pipe(void) {
+	uv_w_handle_t *data = calloc(1, sizeof(uv_w_handle_t));
+	if (data != NULL) {
+		data->cb_close = Val_unit;
+		caml_register_global_root(&(data->cb_close));
+	}
+	return data;
+}
+
 static void unalloc_data(uv_w_handle_t *data) {
 	caml_remove_global_root(&(data->cb_close));
 	caml_remove_global_root(&(data->u.all.cb1));
@@ -1104,11 +1118,11 @@ CAMLprim value w_spawn(value loop, value cb, value file, value args, value env,
 			}
 		} else {
 			stdio_u[i].flags = UV_CREATE_PIPE;
-			// TODO: probably need to give a stream in data.stream?
 			if (Bool_val(Field(stdio_entry, 0)))
-				stdio_u[i].flags = UV_READABLE_PIPE;
+				stdio_u[i].flags |= UV_READABLE_PIPE;
 			if (Bool_val(Field(stdio_entry, 1)))
-				stdio_u[i].flags = UV_WRITABLE_PIPE;
+				stdio_u[i].flags |= UV_WRITABLE_PIPE;
+			stdio_u[i].data.stream = Stream_val(Field(stdio_entry, 2));
 		}
 	}
 	uv_process_options_t options = {
@@ -1142,3 +1156,15 @@ CAMLprim value w_process_get_pid(value handle) {
 	CAMLparam1(handle);
 	CAMLreturn(Val_int(Process_val(handle)->pid));
 }
+
+// ------------- PIPES ----------------------------------------------
+
+CAMLprim value w_pipe_init(value loop, value ipc) {
+	CAMLparam2(loop, ipc);
+	UV_ALLOC_CHECK(handle, uv_pipe_t);
+	UV_ERROR_CHECK_C(uv_pipe_init(Loop_val(loop), Pipe_val(handle), Bool_val(ipc)), free(Pipe_val(handle)));
+	UV_HANDLE_DATA(Pipe_val(handle)) = alloc_data_pipe();
+	if (UV_HANDLE_DATA(Pipe_val(handle)) == NULL)
+		UV_ERROR(0);
+	UV_SUCCESS(handle);
+}

src/macro/eval/evalHash.ml

@@ -155,3 +155,4 @@ let key_nusys_net_SocketAddress = hash "nusys.net.SocketAddress"
 let key_eval_uv_Timer = hash "eval.uv.Timer"
 let key_eval_uv_Process = hash "eval.uv.Process"
 let key_sys_FileWatcherEvent = hash "sys.FileWatcherEvent"
+let key_eval_uv_Pipe = hash "eval.uv.Pipe"

src/macro/eval/evalStdLib.ml

@@ -3772,6 +3772,40 @@ module StdUv = struct
 		let unref = wrap_unref this
 	end
 
+	module Pipe = struct
+		let this vthis = match vthis with
+			| VInstance {ikind = IUv (UvPipe t)} -> t
+			| v -> unexpected_value v "UvPipe"
+		let new_ = (fun vl ->
+			match vl with
+				| [] ->
+					let pipe = wrap_sync (Uv.pipe_init (loop ()) false) in
+					encode_instance key_eval_uv_Pipe ~kind:(IUv (UvPipe pipe))
+				| _ -> assert false
+		)
+		let write = vifun2 (fun vthis data cb ->
+			let this = this vthis in
+			let data = decode_bytes data in
+			wrap_sync (Uv.write this data (wrap_cb_unit cb));
+			vnull
+		)
+		let startRead = vifun1 (fun vthis cb ->
+			let this = this vthis in
+			wrap_sync (Uv.read_start this (wrap_cb cb encode_bytes));
+			vnull
+		)
+		let stopRead = vifun0 (fun vthis ->
+			let this = this vthis in
+			wrap_sync (Uv.read_stop this);
+			vnull
+		)
+		let close = vifun1 (fun vthis cb ->
+			let this = this vthis in
+			wrap_sync (Uv.close this (wrap_cb_unit cb));
+			vnull
+		)
+	end
+
 	module Process = struct
 		let this vthis = match vthis with
 			| VInstance {ikind = IUv (UvProcess t)} -> t
@@ -3784,13 +3818,14 @@ module StdUv = struct
 					let env = Array.of_list (List.map decode_string (decode_array env)) in
 					let cwd = decode_string cwd in
 					let flags = decode_int flags in
-					let stdio = Array.of_list (List.map (function
-							| VEnumValue {eindex = 0; eargs = [|readable; writable|]} ->
+					let stdio = Array.of_list (List.map (fun stdio -> match (decode_enum stdio) with
+							| 0, [readable; writable; pipe] ->
 								let readable = decode_bool readable in
 								let writable = decode_bool writable in
-								Uv.UvIoPipe (readable, writable)
-							| VEnumValue {eindex = 1} -> Uv.UvIoIgnore
-							| VEnumValue {eindex = 2} -> Uv.UvIoInherit
+								let pipe = Pipe.this pipe in
+								Uv.UvIoPipe (readable, writable, pipe)
+							| 1, [] -> Uv.UvIoIgnore
+							| 2, [] -> Uv.UvIoInherit
 							| _ -> assert false
 						) (decode_array stdio)) in
 					let uid = decode_int uid in
@@ -4005,7 +4040,8 @@ let init_constructors builtins =
 	add key_eval_uv_Socket StdUv.Socket.new_;
 	add key_eval_uv_UdpSocket StdUv.UdpSocket.new_;
 	add key_eval_uv_Timer StdUv.Timer.new_;
-	add key_eval_uv_Process StdUv.Process.new_
+	add key_eval_uv_Process StdUv.Process.new_;
+	add key_eval_uv_Pipe StdUv.Pipe.new_
 
 let init_empty_constructors builtins =
 	let h = builtins.empty_constructor_builtins in
@@ -4567,4 +4603,10 @@ let init_standard_library builtins =
 		"getPid",StdUv.Process.getPid;
 		"ref",StdUv.Process.ref_;
 		"unref",StdUv.Process.unref;
+	];
+	init_fields builtins (["eval";"uv"],"Pipe") [] [
+		"write",StdUv.Pipe.write;
+		"startRead",StdUv.Pipe.startRead;
+		"stopRead",StdUv.Pipe.stopRead;
+		"close",StdUv.Pipe.close;
 	]

src/macro/eval/evalValue.ml

@@ -103,6 +103,7 @@ type vuv_value =
 	| UvUdp of Uv.t_udp
 	| UvTimer of Uv.t_timer
 	| UvProcess of Uv.t_process
+	| UvPipe of Uv.t_pipe
 
 type value =
 	| VNull