Merge pull request #5940 from jamessan/master

job: Consume content from rbuffer before invoking the callback again

src/nvim/eval.c

@@ -23183,11 +23183,10 @@ static void on_job_output(Stream *stream, TerminalJobData *data, RBuffer *buf,
     terminal_receive(data->term, ptr, count);
   }
 
+  rbuffer_consumed(buf, count);
   if (callback->type != kCallbackNone) {
     process_job_event(data, callback, type, ptr, count, 0);
   }
-
-  rbuffer_consumed(buf, count);
 }
 
 static void eval_job_process_exit_cb(Process *proc, int status, void *d)

test/functional/core/job_spec.lua

@@ -312,6 +312,24 @@ describe('jobs', function()
     end)
   end)
 
+  it('does not repeat output with slow output handlers', function()
+    source([[
+      let d = {'data': []}
+      function! d.on_stdout(job, data, event) dict
+        call add(self.data, a:data)
+        sleep 200m
+      endfunction
+      if has('win32')
+        let cmd = '1,2,3,4,5 | foreach-object -process {echo $_; sleep 0.1}'
+      else
+        let cmd = ['sh', '-c', 'for i in $(seq 1 5); do echo $i; sleep 0.1; done']
+      endif
+      call jobwait([jobstart(cmd, d)])
+      call rpcnotify(g:channel, 'data', d.data)
+    ]])
+    eq({'notification', 'data', {{{'1', ''}, {'2', ''}, {'3', ''}, {'4', ''}, {'5', ''}}}}, next_msg())
+  end)
+
   describe('jobwait', function()
     it('returns a list of status codes', function()
       source([[