pipes

Author: timbertson

std/async/process.kk

@@ -216,9 +216,10 @@ pub fun run(
     // (which the success path will wait for)
     finalize-process(process, kill-strategy)
 
+  println("TODO: close half of pipe belonging to child...")
   // close the half of any pipes that are only used by the child
-  close-owned(uv-stdout)
-  close-owned(uv-stderr)
+  // close-owned(uv-stdout)
+  // close-owned(uv-stderr)
 
   val result = action(process)
 
@@ -230,8 +231,7 @@ pub fun run(
 
 fun close-owned(stream: stdio-stream): asyncx ()
   match stream
-    Stream-uv(s) ->
-      s.close()
+    Stream-uv(s) -> s.close()
     _ -> ()
 
 pub fun await(process: process): <io,async> exit

test/std/async/process-test.kk

@@ -23,12 +23,12 @@ fun expect-process-finished(process)
 fun suite()
   effectful-test("success")
     expect(())
-      command("true", []).run()
+      command("true", []).run-unit()
     wait(0.2)
 
   effectful-test("failure")
     expect-error-message("Command failed (status:1)")
-      command("false", []).run()
+      command("false", []).run-unit()
     wait(0.2)
 
   effectful-test("allowable failure")
@@ -52,8 +52,8 @@ fun suite()
     expect-process-finished(process.unjust)
     wait(0.2)
 
-  effectful-test("capture output")
-    expect("echo 1 2 words with     many spaces\n") {
-      val c = command("echo", ["1", "2", "words with     many spaces"], output=capture-stdout)
-      c.run(fn(p) p.output)
-    }
+  // effectful-test("capture output")
+  //   expect("echo 1 2 words with     many spaces\n") {
+  //     val c = command("echo", ["1", "2", "words with     many spaces"], output=capture-stdout)
+  //     c.run(fn(p) p.output)
+  //   }

uv/inline/process.c

@@ -88,7 +88,6 @@ static uv_stdio_container_t kk_convert_to_stdio_container(
   } else if (kk_uv_process_is_stream_uv(stream, _ctx)) {
     result.flags = UV_INHERIT_STREAM;
     struct kk_uv_process_Stream_uv* stream_uv = kk_uv_process__as_Stream_uv(stream, _ctx);
-    // TODO: do we need to dup this stream?
     result.data.stream = kk_owned_handle_to_uv_handle(uv_stream_t, stream_uv->value);
   } else {
     kk_fatal_error(EINVAL, "unknown stream type\n");
@@ -187,8 +186,8 @@ static kk_std_core_exn__error kk_uv_proc_pipe(kk_context_t* _ctx) {
   memcpy(writable_p, &writable, sizeof(uv_pipe_t));
 
   kk_std_core_types__tuple2 result = kk_std_core_types__new_Tuple2(
-    uv_handle_to_owned_kk_handle_box(readable_p, kk_handle_free, stream, stream),
-    uv_handle_to_owned_kk_handle_box(writable_p, kk_handle_free, stream, stream),
+    uv_handle_to_owned_kk_handle_box(readable_p, kk_free_fun, stream, stream),
+    uv_handle_to_owned_kk_handle_box(writable_p, kk_free_fun, stream, stream),
     _ctx);
 
   return kk_std_core_exn__new_Ok(kk_std_core_types__tuple2_box(result, _ctx), _ctx);