better pipes and streams

libs/uv/uv.ml

@@ -174,6 +174,7 @@ external listen : 'a -> int -> unit_cb -> unit uv_result = "w_listen"
 external write : 'a -> bytes -> unit_cb -> unit uv_result = "w_write"
 external read_start : 'a -> stream_bytes_cb -> unit uv_result = "w_read_start"
 external read_stop : 'a -> unit uv_result = "w_read_stop"
+external stream_of_handle : 'a -> t_stream = "w_stream_of_handle"
 
 (* ------------- TCP ------------------------------------------------ *)
 
@@ -245,7 +246,7 @@ 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 * t_pipe
+	| UvIoPipe of bool * bool * t_stream
 	| UvIoIgnore
 	| UvIoInherit
 
@@ -256,3 +257,6 @@ 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"
+external pipe_accept : t_loop -> t_pipe -> t_pipe uv_result = "w_pipe_accept"
+external pipe_bind_ipc : t_pipe -> string -> unit uv_result = "w_pipe_bind_ipc"
+external pipe_connect_ipc : t_pipe -> string -> unit_cb -> unit uv_result = "w_pipe_connect_ipc"

libs/uv/uv_stubs.c

@@ -662,6 +662,11 @@ CAMLprim value w_read_stop(value stream) {
 	UV_SUCCESS_UNIT;
 }
 
+CAMLprim value w_stream_of_handle(value handle) {
+	CAMLparam1(handle);
+	CAMLreturn(handle);
+}
+
 // ------------- NETWORK MACROS -------------------------------------
 
 #define UV_SOCKADDR_IPV4(var, host, port) \
@@ -1168,3 +1173,29 @@ CAMLprim value w_pipe_init(value loop, value ipc) {
 		UV_ERROR(0);
 	UV_SUCCESS(handle);
 }
+
+CAMLprim value w_pipe_accept(value loop, value server) {
+	CAMLparam2(loop, server);
+	UV_ALLOC_CHECK(client, uv_pipe_t);
+	UV_ERROR_CHECK_C(uv_pipe_init(Loop_val(loop), Pipe_val(client), 0), free(Pipe_val(client)));
+	UV_HANDLE_DATA(Pipe_val(client)) = alloc_data_pipe();
+	if (UV_HANDLE_DATA(Pipe_val(client)) == NULL)
+		UV_ERROR(0);
+	UV_ERROR_CHECK_C(uv_accept(Stream_val(server), Stream_val(client)), free(Pipe_val(client)));
+	UV_SUCCESS(client);
+}
+
+CAMLprim value w_pipe_bind_ipc(value handle, value path) {
+	CAMLparam2(handle, path);
+	UV_ERROR_CHECK(uv_pipe_bind(Pipe_val(handle), String_val(path)));
+	UV_SUCCESS_UNIT;
+}
+
+CAMLprim value w_pipe_connect_ipc(value handle, value path, value cb) {
+	CAMLparam3(handle, path, cb);
+	UV_ALLOC_CHECK(req, uv_connect_t);
+	UV_REQ_DATA(Connect_val(req)) = (void *)cb;
+	caml_register_global_root(UV_REQ_DATA_A(Connect_val(req)));
+	uv_pipe_connect(Connect_val(req), Pipe_val(handle), String_val(path), (void (*)(uv_connect_t *, int))handle_stream_cb);
+	UV_SUCCESS_UNIT;
+}

src/macro/eval/evalHash.ml

@@ -156,3 +156,4 @@ 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"
+let key_eval_uv_Stream = hash "eval.uv.Stream"

src/macro/eval/evalStdLib.ml

@@ -3476,12 +3476,6 @@ module StdUv = struct
 			let s = wrap_sync (Uv.tcp_init (loop ())) in
 			encode_instance key_eval_uv_Socket ~kind:(IUv (UvTcp s))
 		)
-		let listen = vifun2 (fun vthis backlog cb ->
-			let this = this vthis in
-			let backlog = decode_int backlog in
-			wrap_sync (Uv.listen this backlog (wrap_cb_unit cb));
-			vnull;
-		)
 		let accept = vifun0 (fun vthis ->
 			let this = this vthis in
 			let res = wrap_sync (Uv.tcp_accept (loop ()) this) in
@@ -3515,36 +3509,6 @@ module StdUv = struct
 				| _ -> assert false);
 			vnull
 		)
-		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 end_ = vifun1 (fun vthis cb ->
-			let this = this vthis in
-			wrap_sync (Uv.shutdown this (fun res ->
-				match res with
-					| Uv.UvError err -> ignore (call_value cb [wrap_error err; vnull])
-					| Uv.UvSuccess () -> wrap_sync (Uv.close this (wrap_cb_unit cb))
-				));
-			vnull
-		)
-		let close = vifun1 (fun vthis cb ->
-			let this = this vthis in
-			wrap_sync (Uv.close this (wrap_cb_unit cb));
-			vnull
-		)
 		let setKeepAlive = vifun2 (fun vthis enable initialDelay ->
 			let this = this vthis in
 			let enable = decode_bool enable in
@@ -3569,8 +3533,11 @@ module StdUv = struct
 		)
 		let getSockName = getName Uv.tcp_getsockname
 		let getPeerName = getName Uv.tcp_getpeername
-		let ref_ = wrap_ref this
-		let unref = wrap_unref this
+		let asStream = vifun0 (fun vthis ->
+			let this = this vthis in
+			let stream = Uv.stream_of_handle this in
+			encode_instance key_eval_uv_Stream ~kind:(IUv (UvStream stream))
+		)
 	end
 
 	module UdpSocket = struct
@@ -3783,12 +3750,49 @@ module StdUv = struct
 					encode_instance key_eval_uv_Pipe ~kind:(IUv (UvPipe pipe))
 				| _ -> assert false
 		)
+		let accept = vifun0 (fun vthis ->
+			let this = this vthis in
+			let res = wrap_sync (Uv.pipe_accept (loop ()) this) in
+			encode_instance key_eval_uv_Pipe ~kind:(IUv (UvPipe res))
+		)
+		let bindIpc = vifun1 (fun vthis path ->
+			let this = this vthis in
+			let path = decode_string path in
+			Uv.pipe_bind_ipc this path;
+			vnull
+		)
+		let connectIpc = vifun2 (fun vthis path cb ->
+			let this = this vthis in
+			let path = decode_string path in
+			wrap_sync (Uv.pipe_connect_ipc this path (wrap_cb_unit cb));
+			vnull
+		)
+		let asStream = vifun0 (fun vthis ->
+			let this = this vthis in
+			let stream = Uv.stream_of_handle this in
+			encode_instance key_eval_uv_Stream ~kind:(IUv (UvStream stream))
+		)
+	end
+
+	module Stream = struct
+		let this vthis = match vthis with
+			| VInstance {ikind = IUv (UvStream t)} -> t
+			| v -> unexpected_value v "UvStream"
 		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 end_ = vifun1 (fun vthis cb ->
+			let this = this vthis in
+			wrap_sync (Uv.shutdown this (fun res ->
+				match res with
+					| Uv.UvError err -> ignore (call_value cb [wrap_error err; vnull])
+					| Uv.UvSuccess () -> wrap_sync (Uv.close this (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));
@@ -3799,11 +3803,19 @@ module StdUv = struct
 			wrap_sync (Uv.read_stop this);
 			vnull
 		)
+		let listen = vifun2 (fun vthis backlog cb ->
+			let this = this vthis in
+			let backlog = decode_int backlog in
+			wrap_sync (Uv.listen this backlog (wrap_cb_unit cb));
+			vnull;
+		)
 		let close = vifun1 (fun vthis cb ->
 			let this = this vthis in
 			wrap_sync (Uv.close this (wrap_cb_unit cb));
 			vnull
 		)
+		let ref_ = wrap_ref this
+		let unref = wrap_unref this
 	end
 
 	module Process = struct
@@ -3822,7 +3834,7 @@ module StdUv = struct
 							| 0, [readable; writable; pipe] ->
 								let readable = decode_bool readable in
 								let writable = decode_bool writable in
-								let pipe = Pipe.this pipe in
+								let pipe = Stream.this pipe in
 								Uv.UvIoPipe (readable, writable, pipe)
 							| 1, [] -> Uv.UvIoIgnore
 							| 2, [] -> Uv.UvIoInherit
@@ -4561,19 +4573,12 @@ let init_standard_library builtins =
 	init_fields builtins (["eval";"uv"],"Socket") [] [
 		"bindTcp",StdUv.Socket.bindTcp;
 		"connectTcp",StdUv.Socket.connectTcp;
-		"listen",StdUv.Socket.listen;
 		"accept",StdUv.Socket.accept;
-		"write",StdUv.Socket.write;
-		"startRead",StdUv.Socket.startRead;
-		"stopRead",StdUv.Socket.stopRead;
-		"end",StdUv.Socket.end_;
-		"close",StdUv.Socket.close;
 		"setKeepAlive",StdUv.Socket.setKeepAlive;
 		"setNoDelay",StdUv.Socket.setNoDelay;
 		"getSockName",StdUv.Socket.getSockName;
 		"getPeerName",StdUv.Socket.getPeerName;
-		"ref",StdUv.Socket.ref_;
-		"unref",StdUv.Socket.unref;
+		"asStream",StdUv.Socket.asStream;
 	];
 	init_fields builtins (["eval";"uv"],"UdpSocket") [] [
 		"addMembership",StdUv.UdpSocket.addMembership;
@@ -4611,8 +4616,18 @@ let init_standard_library builtins =
 		"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;
+		"accept",StdUv.Pipe.accept;
+		"bindIpc",StdUv.Pipe.bindIpc;
+		"connectIpc",StdUv.Pipe.connectIpc;
+		"asStream",StdUv.Pipe.asStream;
+	];
+	init_fields builtins (["eval";"uv"],"Stream") [] [
+		"listen",StdUv.Stream.listen;
+		"write",StdUv.Stream.write;
+		"startRead",StdUv.Stream.startRead;
+		"stopRead",StdUv.Stream.stopRead;
+		"end",StdUv.Stream.end_;
+		"close",StdUv.Stream.close;
+		"ref",StdUv.Stream.ref_;
+		"unref",StdUv.Stream.unref;
 	]

src/macro/eval/evalValue.ml

@@ -99,6 +99,7 @@ type vuv_value =
 	| UvFsEvent of Uv.t_fs_event
 	| UvStat of Uv.t_stat
 	| UvDirent of (string * int)
+	| UvStream of Uv.t_stream
 	| UvTcp of Uv.t_tcp
 	| UvUdp of Uv.t_udp
 	| UvTimer of Uv.t_timer