eval: Fix buffering of data in job autocommands

Job autocommands will no longer buffer data chunks that don't end in newlines
characters.

src/nvim/eval.c

@@ -454,8 +454,8 @@ static dictitem_T vimvars_var;                  /* variable used for v: */
 // Memory pool for reusing JobEvent structures
 typedef struct {
   int id;
-  char *name, *type, *received;
+  char *name, *type;
-  size_t received_len;
+  list_T *received;
 } JobEvent;
 #define JobEventFreer(x)
 KMEMPOOL_INIT(JobEventPool, JobEvent, JobEventFreer)
@@ -19705,72 +19705,73 @@ char_u *do_string_sub(char_u *str, char_u *pat, char_u *sub, char_u *flags)
 
 // JobActivity autocommands will execute vimscript code, so it must be executed
 // on Nvim main loop
-#define push_job_event(j, r, t, eof)                                 \
+static inline void push_job_event(Job *job, RStream *rstream, char *type)
-  do {                                                               \
+{
-    JobEvent *event_data = kmp_alloc(JobEventPool, job_event_pool);  \
+  JobEvent *event_data = kmp_alloc(JobEventPool, job_event_pool);
-    event_data->received = NULL;                                     \
+  event_data->received = NULL;
-    size_t read_count = 0;                                           \
+  if (rstream) {
-    if (r) {                                                         \
+    event_data->received = list_alloc();
-      if (eof) {                                                     \
+    char *ptr = rstream_read_ptr(rstream);
-        read_count = rstream_pending(r);                             \
+    size_t count = rstream_pending(rstream);
-      } else {                                                       \
+    size_t remaining = count;
-        char *read = rstream_read_ptr(r);                            \
+    size_t off = 0;
-        char *lastnl = xmemrchr(read, NL, rstream_pending(r));       \
+
-        if (lastnl) {                                                \
+    while (off < remaining) {
-          read_count = (size_t) (lastnl - read) + 1;                 \
+      // append the line
-        } else if (rstream_available(r) == 0) {                      \
+      if (ptr[off] == NL) {
-          /* No newline or room to grow; flush everything. */        \
+        list_append_string(event_data->received, (uint8_t *)ptr, off);
-          read_count = rstream_pending(r);                           \
+        size_t skip = off + 1;
-        }                                                            \
+        ptr += skip;
-      }                                                              \
+        remaining -= skip;
-      if (read_count == 0) {                                         \
+        off = 0;
-        /* Either we're at EOF or we need to wait until next time */ \
+        continue;
-        /* to receive a '\n. */                                      \
+      }
-        kmp_free(JobEventPool, job_event_pool, event_data);          \
+      if (ptr[off] == NUL) {
-        return;                                                      \
+        // Translate NUL to NL
-      }                                                              \
+        ptr[off] = NL;
-      event_data->received_len = read_count;                         \
+      }
-      event_data->received = xmallocz(read_count);                   \
+      off++;
-      rstream_read(r, event_data->received, read_count);             \
+    }
-    }                                                                \
+    list_append_string(event_data->received, (uint8_t *)ptr, off);
-    event_data->id = job_id(j);                                      \
+    rbuffer_consumed(rstream_buffer(rstream), count);
-    event_data->name = job_data(j);                                  \
+  }
-    event_data->type = t;                                            \
+  event_data->id = job_id(job);
-    event_push((Event) {                                             \
+  event_data->name = job_data(job);
-      .handler = on_job_event,                                       \
+  event_data->type = type;
-      .data = event_data                                             \
+  event_push((Event) {
-    }, true);                                                        \
+    .handler = on_job_event,
-  } while(0)
+    .data = event_data
+  }, true);
+}
 
 static void on_job_stdout(RStream *rstream, void *data, bool eof)
 {
-  if (rstream_pending(rstream)) {
+  if (!eof) {
-    push_job_event(data, rstream, "stdout", eof);
+    push_job_event(data, rstream, "stdout");
   }
 }
 
 static void on_job_stderr(RStream *rstream, void *data, bool eof)
 {
-  if (rstream_pending(rstream)) {
+  if (!eof) {
-    push_job_event(data, rstream, "stderr", eof);
+    push_job_event(data, rstream, "stderr");
   }
 }
 
 static void on_job_exit(Job *job, void *data)
 {
-  push_job_event(job, NULL, "exit", true);
+  push_job_event(job, NULL, "exit");
 }
 
 static void on_job_event(Event event)
 {
   JobEvent *data = event.data;
-  apply_job_autocmds(data->id, data->name, data->type,
+  apply_job_autocmds(data->id, data->name, data->type, data->received);
-                     data->received, data->received_len);
   kmp_free(JobEventPool, job_event_pool, data);
 }
 
 static void apply_job_autocmds(int id, char *name, char *type,
-                               char *received, size_t received_len)
+                               list_T *received)
 {
   // Create the list which will be set to v:job_data
   list_T *list = list_alloc();
@@ -19781,12 +19782,9 @@ static void apply_job_autocmds(int id, char *name, char *type,
     listitem_T *str_slot = listitem_alloc();
     str_slot->li_tv.v_type = VAR_LIST;
     str_slot->li_tv.v_lock = 0;
-    str_slot->li_tv.vval.v_list =
+    str_slot->li_tv.vval.v_list = received;
-      string_to_list((char_u *) received, received_len, false);
     str_slot->li_tv.vval.v_list->lv_refcount++;
     list_append(list, str_slot);
-
-    free(received);
   }
 
   // Update v:job_data for the autocommands

test/functional/job/job_spec.lua

@@ -41,11 +41,11 @@ describe('jobs', function()
     nvim('command', "let j = jobstart('xxx', 'cat', ['-'])")
     neq(0, eval('j'))
     nvim('command', 'call jobsend(j, "abc\\n")')
-    eq({'notification', 'stdout', {{'abc'}}}, next_message())
+    eq({'notification', 'stdout', {{'abc', ''}}}, next_message())
     nvim('command', 'call jobsend(j, "123\\nxyz\\n")')
-    eq({'notification', 'stdout', {{'123', 'xyz'}}}, next_message())
+    eq({'notification', 'stdout', {{'123', 'xyz', ''}}}, next_message())
     nvim('command', 'call jobsend(j, [123, "xyz"])')
-    eq({'notification', 'stdout', {{'123', 'xyz'}}}, next_message())
+    eq({'notification', 'stdout', {{'123', 'xyz', ''}}}, next_message())
     nvim('command', "call jobstop(j)")
     eq({'notification', 'exit', {0}}, next_message())
   end)
@@ -60,24 +60,41 @@ describe('jobs', function()
     -- v:job_data preserves NULs.
     nvim('command', notify_str('v:job_data[1]', 'get(v:job_data, 2)'))
     nvim('command', "let j = jobstart('xxx', 'cat', ['"..filename.."'])")
-    eq({'notification', 'stdout', {{'abc\ndef'}}}, next_message())
+    eq({'notification', 'stdout', {{'abc\ndef', ''}}}, next_message())
     eq({'notification', 'exit', {0}}, next_message())
     os.remove(filename)
 
     -- jobsend() preserves NULs.
     nvim('command', "let j = jobstart('xxx', 'cat', ['-'])")
     nvim('command', [[call jobsend(j, ["123\n456"])]])
-    eq({'notification', 'stdout', {{'123\n456'}}}, next_message())
+    eq({'notification', 'stdout', {{'123\n456', ''}}}, next_message())
     nvim('command', "call jobstop(j)")
   end)
 
-  it('will hold data if it does not end in a newline', function()
+  it('will not buffer data if it doesnt end in newlines', function()
     nvim('command', notify_str('v:job_data[1]', 'get(v:job_data, 2)'))
     nvim('command', "let j = jobstart('xxx', 'cat', ['-'])")
     nvim('command', 'call jobsend(j, "abc\\nxyz")')
-    eq({'notification', 'stdout', {{'abc'}}}, next_message())
+    eq({'notification', 'stdout', {{'abc', 'xyz'}}}, next_message())
+    nvim('command', "call jobstop(j)")
+    eq({'notification', 'exit', {0}}, next_message())
+  end)
+
+  it('can preserve newlines', function()
+    nvim('command', notify_str('v:job_data[1]', 'get(v:job_data, 2)'))
+    nvim('command', "let j = jobstart('xxx', 'cat', ['-'])")
+    nvim('command', 'call jobsend(j, "a\\n\\nc\\n\\n\\n\\nb\\n\\n")')
+    eq({'notification', 'stdout', {{'a', '', 'c', '', '', '', 'b', '', ''}}},
+      next_message())
+  end)
+
+  it('can preserve nuls', function()
+    nvim('command', notify_str('v:job_data[1]', 'get(v:job_data, 2)'))
+    nvim('command', "let j = jobstart('xxx', 'cat', ['-'])")
+    nvim('command', 'call jobsend(j, ["\n123\n", "abc\\nxyz\n"])')
+    eq({'notification', 'stdout', {{'\n123\n', 'abc\nxyz\n', ''}}},
+      next_message())
     nvim('command', "call jobstop(j)")
-    eq({'notification', 'stdout', {{'xyz'}}}, next_message())
     eq({'notification', 'exit', {0}}, next_message())
   end)