Merge PR #3029 'Refactor event processing architecture'

Helped-by: oni-link <knil.ino@gmail.com>
Reviewed-by: oni-link <knil.ino@gmail.com>

1 files changed, 20 insertions, 5 deletions

diff --git a/src/nvim/os/shell.c b/src/nvim/os/shell.c
index b9c5db4261..2d97c4bf4f 100644
--- a/src/nvim/os/shell.c
+++ b/src/nvim/os/shell.c
@@ -205,13 +205,15 @@ static int do_os_system(char **argv,
   xstrlcpy(prog, argv[0], MAXPATHL);
 
   Stream in, out, err;
-  UvProcess uvproc = uv_process_init(&buf);
+  UvProcess uvproc = uv_process_init(&loop, &buf);
   Process *proc = &uvproc.process;
+  Queue *events = queue_new_child(loop.events);
+  proc->events = events;
   proc->argv = argv;
   proc->in = input != NULL ? &in : NULL;
   proc->out = &out;
   proc->err = &err;
-  if (!process_spawn(&loop, proc)) {
+  if (!process_spawn(proc)) {
     loop_poll_events(&loop, 0);
     // Failed, probably due to `sh` not being executable
     if (!silent) {
@@ -219,14 +221,22 @@ static int do_os_system(char **argv,
       msg_outtrans((char_u *)prog);
       msg_putchar('\n');
     }
+    queue_free(events);
     return -1;
   }
 
+  // We want to deal with stream events as fast a possible while queueing
+  // process events, so reset everything to NULL. It prevents closing the
+  // streams while there's still data in the OS buffer(due to the process
+  // exiting before all data is read).
   if (input != NULL) {
+    proc->in->events = NULL;
     wstream_init(proc->in, 0);
   }
+  proc->out->events = NULL;
   rstream_init(proc->out, 0);
   rstream_start(proc->out, data_cb);
+  proc->err->events = NULL;
   rstream_init(proc->err, 0);
   rstream_start(proc->err, data_cb);
 
@@ -247,7 +257,7 @@ static int do_os_system(char **argv,
   // the UI
   ui_busy_start();
   ui_flush();
-  int status = process_wait(proc, -1);
+  int status = process_wait(proc, -1, NULL);
   ui_busy_stop();
 
   // prepare the out parameters if requested
@@ -267,6 +277,9 @@ static int do_os_system(char **argv,
     }
   }
 
+  assert(queue_empty(events));
+  queue_free(events);
+
   return status;
 }
 
@@ -285,7 +298,8 @@ static void dynamic_buffer_ensure(DynamicBuffer *buf, size_t desired)
   buf->data = xrealloc(buf->data, buf->cap);
 }
 
-static void system_data_cb(Stream *stream, RBuffer *buf, void *data, bool eof)
+static void system_data_cb(Stream *stream, RBuffer *buf, size_t count,
+    void *data, bool eof)
 {
   DynamicBuffer *dbuf = data;
 
@@ -295,7 +309,8 @@ static void system_data_cb(Stream *stream, RBuffer *buf, void *data, bool eof)
   dbuf->len += nread;
 }
 
-static void out_data_cb(Stream *stream, RBuffer *buf, void *data, bool eof)
+static void out_data_cb(Stream *stream, RBuffer *buf, size_t count, void *data,
+    bool eof)
 {
   size_t cnt;
   char *ptr = rbuffer_read_ptr(buf, &cnt);