Merge #7646 from bfredl/chan_buffered

Document and defer error message when buffered stream would overwrite channels dict key

runtime/autoload/provider.vim

@@ -1,20 +0,0 @@
-" Common functionality for providers
-
-let s:stderr = {}
-
-function! provider#stderr_collector(chan_id, data, event)
-   let stderr = get(s:stderr, a:chan_id, [''])
-   let stderr[-1] .= a:data[0]
-   call extend(stderr, a:data[1:])
-   let s:stderr[a:chan_id] = stderr
-endfunction
-
-function! provider#clear_stderr(chan_id)
-  if has_key(s:stderr, a:chan_id)
-    call remove(s:stderr, a:chan_id)
-  endif
-endfunction
-
-function! provider#get_stderr(chan_id)
-   return get(s:stderr, a:chan_id, [])
-endfunction

runtime/autoload/provider/clipboard.vim

@@ -7,7 +7,7 @@ let s:clipboard = {}
 
 " When caching is enabled, store the jobid of the xclip/xsel process keeping
 " ownership of the selection, so we know how long the cache is valid.
-let s:selection = { 'owner': 0, 'data': [], 'on_stderr': function('provider#stderr_collector') }
+let s:selection = { 'owner': 0, 'data': [], 'stderr_buffered': v:true }
 
 function! s:selection.on_exit(jobid, data, event) abort
   " At this point this nvim instance might already have launched
@@ -16,12 +16,10 @@ function! s:selection.on_exit(jobid, data, event) abort
     let self.owner = 0
   endif
   if a:data != 0
-    let stderr = provider#get_stderr(a:jobid)
     echohl WarningMsg
-    echomsg 'clipboard: error invoking '.get(self.argv, 0, '?').': '.join(stderr)
+    echomsg 'clipboard: error invoking '.get(self.argv, 0, '?').': '.join(self.stderr)
     echohl None
   endif
-  call provider#clear_stderr(a:jobid)
 endfunction
 
 let s:selections = { '*': s:selection, '+': copy(s:selection) }
@@ -142,12 +140,13 @@ function! s:clipboard.set(lines, regtype, reg) abort
     return 0
   end
 
-  let selection = s:selections[a:reg]
-  if selection.owner > 0
+  if s:selections[a:reg].owner > 0
     " The previous provider instance should exit when the new one takes
     " ownership, but kill it to be sure we don't fill up the job table.
-    call jobstop(selection.owner)
+    call jobstop(s:selections[a:reg].owner)
   end
+  let s:selections[a:reg] = copy(s:selection)
+  let selection = s:selections[a:reg]
   let selection.data = [a:lines, a:regtype]
   let argv = split(s:copy[a:reg], " ")
   let selection.argv = argv

runtime/autoload/provider/node.vim

@@ -3,7 +3,7 @@ if exists('g:loaded_node_provider')
 endif
 let g:loaded_node_provider = 1
 
-let s:job_opts = {'rpc': v:true, 'on_stderr': function('provider#stderr_collector')}
+let s:job_opts = {'rpc': v:true, 'stderr_buffered': v:true}
 
 function! s:is_minimum_version(version, min_major, min_minor) abort
   if empty(a:version)
@@ -73,19 +73,18 @@ function! provider#node#Require(host) abort
   call add(args, provider#node#Prog())
 
   try
-    let channel_id = jobstart(args, s:job_opts)
+    let job = copy(s:job_opts)
+    let channel_id = jobstart(args, job)
     if rpcrequest(channel_id, 'poll') ==# 'ok'
       return channel_id
     endif
   catch
     echomsg v:throwpoint
     echomsg v:exception
-    for row in provider#get_stderr(channel_id)
+    for row in job.stderr
       echomsg row
     endfor
   endtry
-  finally
-    call provider#clear_stderr(channel_id)
   endtry
   throw remote#host#LoadErrorForHost(a:host.orig_name, '$NVIM_NODE_LOG_FILE')
 endfunction

runtime/autoload/provider/pythonx.vim

@@ -5,7 +5,7 @@ endif
 
 let s:loaded_pythonx_provider = 1
 
-let s:job_opts = {'rpc': v:true, 'on_stderr': function('provider#stderr_collector')}
+let s:job_opts = {'rpc': v:true, 'stderr_buffered': v:true}
 
 function! provider#pythonx#Require(host) abort
   let ver = (a:host.orig_name ==# 'python') ? 2 : 3
@@ -21,18 +21,17 @@ function! provider#pythonx#Require(host) abort
   endfor
 
   try
-    let channel_id = jobstart(args, s:job_opts)
+    let job = copy(s:job_opts)
+    let channel_id = jobstart(args, job)
     if rpcrequest(channel_id, 'poll') ==# 'ok'
       return channel_id
     endif
   catch
     echomsg v:throwpoint
     echomsg v:exception
-    for row in provider#get_stderr(channel_id)
+    for row in job.stderr
       echomsg row
     endfor
-  finally
-    call provider#clear_stderr(channel_id)
   endtry
   throw remote#host#LoadErrorForHost(a:host.orig_name,
         \ '$NVIM_PYTHON_LOG_FILE')

runtime/doc/channel.txt

@@ -43,7 +43,7 @@ bytes. Additionally, for a job channel using rpc, bytes can still be
 read over its stderr. Similarily, only bytes can be written to nvim's own stderr.
 
 						    *channel-callback* *buffered*
-					*on_stdout* *on_stderr* *on_stdin* *on_data*
+				*E5210* *on_stdout* *on_stderr* *on_stdin* *on_data*
 A callback function `on_{stream}` will be invoked with data read from the
 channel.  By default, the callback will be invoked immediately when data is
 available, to facilitate interactive communication. The same callback will
@@ -52,7 +52,11 @@ Alternatively the `{stream}_buffered` option can be set to invoke the callback
 only when the underlying stream reaches EOF, and will then be passed in
 complete output.  This is helpful when only the complete output is useful, and
 not partial data. Futhermore if `{stream}_buffered` is set but not a callback,
-the data is saved in the options dict, with the stream name as key.
+the data is saved in the options dict, with the stream name as key. For this
+to work a new options dict must be used for each opened channel. If a script
+uses a global `s:job_opts` dict, it can be copied with |copy()| before supplying
+it to |jobstart()|. If a dict is reused, so that the dict key already is
+occupied, error `E5210` will be raised.
 
 - The arguments passed to the callback function are:
 

src/nvim/channel.c

@@ -599,6 +599,7 @@ static void on_stdio_input(Stream *stream, RBuffer *buf, size_t count,
   on_channel_output(stream, chan, buf, count, eof, &chan->on_stdout, "stdin");
 }
 
+/// @param type must have static lifetime
 static void on_channel_output(Stream *stream, Channel *chan, RBuffer *buf,
                               size_t count, bool eof, CallbackReader *reader,
                               const char *type)
@@ -613,14 +614,20 @@ static void on_channel_output(Stream *stream, Channel *chan, RBuffer *buf,
       if (reader->cb.type != kCallbackNone) {
         process_channel_event(chan, &reader->cb, type, reader->buffer.ga_data,
                               (size_t)reader->buffer.ga_len, 0);
-        ga_clear(&reader->buffer);
       } else if (reader->self) {
-        list_T *data = buffer_to_tv_list(reader->buffer.ga_data,
-                                         (size_t)reader->buffer.ga_len);
-        tv_dict_add_list(reader->self, type, strlen(type), data);
+        if (tv_dict_find(reader->self, type, -1) == NULL) {
+          list_T *data = buffer_to_tv_list(reader->buffer.ga_data,
+                                           (size_t)reader->buffer.ga_len);
+          tv_dict_add_list(reader->self, type, strlen(type), data);
+        } else {
+            // can't display error message now, defer it.
+            channel_incref(chan);
+            multiqueue_put(chan->events, on_buffered_error, 2, chan, type);
+        }
       } else {
         abort();
       }
+      ga_clear(&reader->buffer);
     } else if (reader->cb.type != kCallbackNone) {
       process_channel_event(chan, &reader->cb, type, ptr, 0, 0);
     }
@@ -641,6 +648,14 @@ static void on_channel_output(Stream *stream, Channel *chan, RBuffer *buf,
   }
 }
 
+static void on_buffered_error(void **args)
+{
+  Channel *chan = (Channel *)args[0];
+  const char *stream = (const char *)args[1];
+  EMSG3(_(e_streamkey), stream, chan->id);
+  channel_decref(chan);
+}
+
 static void channel_process_exit_cb(Process *proc, int status, void *data)
 {
   Channel *chan = data;

src/nvim/globals.h

@@ -1063,6 +1063,9 @@ EXTERN char_u e_stdiochan2[] INIT(= N_(
 EXTERN char_u e_invstream[] INIT(= N_("E906: invalid stream for channel"));
 EXTERN char_u e_invstreamrpc[] INIT(= N_(
     "E906: invalid stream for rpc channel, use 'rpc'"));
+EXTERN char_u e_streamkey[] INIT(= N_(
+    "E5210: dict key '%s' already set for buffered stream in channel %"
+    PRIu64));
 EXTERN char_u e_libcall[] INIT(= N_("E364: Library call failed for \"%s()\""));
 EXTERN char_u e_mkdir[] INIT(= N_("E739: Cannot create directory %s: %s"));
 EXTERN char_u e_markinval[] INIT(= N_("E19: Mark has invalid line number"));

test/functional/core/channels_spec.lua

@@ -246,6 +246,22 @@ describe('channels', function()
     eq({"notification", "exit", {id, 0, {'10 PRINT "NVIM"',
                                          '20 GOTO 10', ''}}}, next_msg())
 
+    -- if dict is reused the new value is not stored,
+    -- but nvim also does not crash
+    command("let id = jobstart(['cat'], g:job_opts)")
+    id = eval("g:id")
+
+    command([[call chansend(id, "cat text\n")]])
+    sleep(10)
+    command("call chanclose(id, 'stdin')")
+
+    -- old value was not overwritten
+    eq({"notification", "exit", {id, 0, {'10 PRINT "NVIM"',
+                                         '20 GOTO 10', ''}}}, next_msg())
+
+    -- and an error was thrown.
+    eq("E5210: dict key 'stdout' already set for buffered stream in channel "..id, eval('v:errmsg'))
+
     -- reset dictionary
     source([[
       let g:job_opts = {
@@ -261,6 +277,5 @@ describe('channels', function()
 
     -- works correctly with no output
     eq({"notification", "exit", {id, 1, {''}}}, next_msg())
-
   end)
 end)