channel: check for existance before trying to set key

This avoids an error message in async context, where it is not safe.

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)