Merge PR #2247 'Refactor/enhance job api'

18 files changed, 900 insertions, 251 deletions

diff --git a/runtime/doc/Makefile b/runtime/doc/Makefile
index 2f9063231d..4bced03d7e 100644
--- a/runtime/doc/Makefile
+++ b/runtime/doc/Makefile
@@ -55,6 +55,7 @@ DOCS = \
 	nvim_intro.txt \
 	nvim_provider.txt \
 	nvim_python.txt \
+	nvim_terminal_emulator.txt \
 	options.txt \
 	os_dos.txt \
 	os_mac.txt \
@@ -176,6 +177,7 @@ HTMLS = \
 	nvim_intro.html \
 	nvim_provider.html \
 	nvim_python.html \
+	nvim_terminal_emulator.html \
 	options.html \
 	os_dos.html \
 	os_mac.html \
diff --git a/runtime/doc/autocmd.txt b/runtime/doc/autocmd.txt
index dc236fc78f..1393131ab7 100644
--- a/runtime/doc/autocmd.txt
+++ b/runtime/doc/autocmd.txt
@@ -253,6 +253,7 @@ Name			triggered by ~
 |BufNew|		just after creating a new buffer
 
 |SwapExists|		detected an existing swap file
+|TermOpen|		when a terminal buffer is starting
 
 	Options
 |FileType|		when the 'filetype' option has been set
@@ -307,7 +308,6 @@ Name			triggered by ~
 |InsertLeave|		when leaving Insert mode
 |InsertCharPre|		when a character was typed in Insert mode, before
 			inserting it
-|JobActivity|		when something interesting happens with a job
 
 |TextChanged|		after a change was made to the text in Normal mode
 |TextChangedI|		after a change was made to the text in Insert mode
@@ -733,10 +733,6 @@ InsertEnter			Just before starting Insert mode.  Also for
 							*InsertLeave*
 InsertLeave			When leaving Insert mode.  Also when using
 				CTRL-O |i_CTRL-O|.  But not for |i_CTRL-C|.
-						{Nvim}  *JobActivity*
-JobActivity			When something interesting happens with a job
-				spawned by |jobstart()|. See |job-control| for
-				details. 
 							*MenuPopup*
 MenuPopup			Just before showing the popup menu (under the
 				right mouse button).  Useful for adjusting the
@@ -876,6 +872,11 @@ TermChanged			After the value of 'term' has changed.  Useful
 				for re-loading the syntax file to update the
 				colors, fonts and other terminal-dependent
 				settings.  Executed for all loaded buffers.
+						 {Nvim} *TermOpen*
+TermOpen			When a terminal buffer is starting.  This can
+				be used to configure the terminal emulator by
+				setting buffer variables.
+				See |nvim-terminal-emulator| for details.
 							*TermResponse*
 TermResponse			After the response to |t_RV| is received from
 				the terminal.  The value of |v:termresponse|
diff --git a/runtime/doc/eval.txt b/runtime/doc/eval.txt
index 1d18a61dbd..5f572e9caa 100644
--- a/runtime/doc/eval.txt
+++ b/runtime/doc/eval.txt
@@ -4012,6 +4012,15 @@ items({dict})						*items()*
 		entry and the value of this entry.  The |List| is in arbitrary
 		order.
 
+jobclose({job}[, {stream}])				{Nvim} *jobclose()*
+		Close {job}'s {stream}, which can be one "stdin", "stdout" or
+		"stderr". If {stream} is omitted, all streams are closed.
+
+jobresize({job}, {width}, {height})			{Nvim} *jobresize()*
+		Resize {job}'s pseudo terminal window to {width} and {height}.
+		This function will fail if used on jobs started without the
+		"pty" option.
+
 jobsend({job}, {data})					{Nvim} *jobsend()*
 		Send data to {job} by writing it to the stdin of the process.
 		Returns 1 if the write succeeded, 0 otherwise.
@@ -4024,14 +4033,28 @@ jobsend({job}, {data})					{Nvim} *jobsend()*
 			:call jobsend(j, ["abc", "123\n456", ""])
 < 		will send "abc<NL>123<NUL>456<NL>".
 
-jobstart({name}, {prog}[, {argv}])			{Nvim} *jobstart()*
-		Spawns {prog} as a job and associate it with the {name} string,
-		which will be used to match the "filename pattern" in
-		|JobActivity| events. It returns:
-		  - The job id on success, which is used by |jobsend()| and
+jobstart({argv}[, {opts}])			{Nvim} *jobstart()*
+		Spawns {argv}(list) as a job. If passed, {opts} must be a
+		dictionary with any of the following keys:
+		  - on_stdout: stdout event handler
+		  - on_stderr: stderr event handler
+		  - on_exit: exit event handler
+		  - pty: If set, the job will be connected to a new pseudo
+		    terminal, and the job streams are connected to the master
+		    file descriptor.
+		  - width: Width of the terminal screen(only if pty is set)
+		  - height: Height of the terminal screen(only if pty is set)
+		  - TERM: $TERM environment variable(only if pty is set)
+		Either funcrefs or function names can be passed as event
+		handlers. The {opts} object is also used as the "self"
+		argument for the callback, so the caller may pass arbitrary
+		data by setting other key.(see |Dictionary-function| for more
+		information).
+		Returns:
+		  - The job ID on success, which is used by |jobsend()| and
 		    |jobstop()|
 		  - 0 when the job table is full or on invalid arguments
-		  - -1 when {prog} is not executable
+		  - -1 when {argv}[0] is not executable
 		See |job-control| for more information.
 
 jobstop({job})						{Nvim} *jobstop()*
@@ -4042,6 +4065,24 @@ jobstop({job})						{Nvim} *jobstop()*
 		`v:job_data[0]` set to `exited`.  See |job-control| for more
 		information.
 
+jobwait({ids}[, {timeout}])				{Nvim} *jobwait()*
+		Wait for a set of jobs to finish. The {ids} argument is a list
+		of ids for jobs that will be waited for. If passed, {timeout}
+		is the maximum number of milliseconds to wait. While this
+		function is executing, callbacks for jobs not in the {ids}
+		list can be executed. Also, the screen wont be updated unless
+		|:redraw| is invoked by one of the callbacks.
+
+		Returns a list of integers with the same length as {ids}, with
+		each integer representing the wait result for the
+		corresponding job id. The possible values for the resulting
+		integers are:
+
+		* the job return code if the job exited
+		* -1 if the wait timed out for the job
+		* -2 if the job was interrupted
+		* -3 if the job id is invalid.
+
 join({list} [, {sep}])					*join()*
 		Join the items in {list} together into one String.
 		When {sep} is specified it is put in between the items.  If
@@ -6277,6 +6318,17 @@ tempname()					*tempname()* *temp-file-name*
 		For MS-Windows forward slashes are used when the 'shellslash'
 		option is set or when 'shellcmdflag' starts with '-'.
 
+termopen({command}[, {opts}])			{Nvim} *termopen()*
+		Spawns {command} using the shell in a new pseudo-terminal
+		session connected to the current buffer.  This function fails
+		if the current buffer is modified (all buffer contents are
+		destroyed).  The {opts} dict is similar to the one passed to
+		|jobstart()|, but the `pty`, `width`, `height`, and `TERM` fields are
+		ignored: `height`/`width` are taken from the current window and
+		$TERM is set to "xterm-256color".  Returns the same values as
+		|jobstart()|.
+
+		See |nvim-terminal-emulator| for more information.
 
 tan({expr})						*tan()*
 		Return the tangent of {expr}, measured in radians, as a |Float|
diff --git a/runtime/doc/job_control.txt b/runtime/doc/job_control.txt
index 1faf9bcd94..dc746bbe99 100644
--- a/runtime/doc/job_control.txt
+++ b/runtime/doc/job_control.txt
@@ -37,37 +37,28 @@ for details
 ==============================================================================
 2. Usage						*job-control-usage*
 
-Here's a quick one-liner that creates a job which invokes the "ls" shell
-command and prints the result:
->
-    call jobstart('', 'ls', ['-a'])|au JobActivity * echo v:job_data|au!
-    JobActivity
-
-In the one-liner above, creating the JobActivity event handler immediately
-after the call to jobstart() is not a race because the Nvim job system will
-not publish the job result (even though it may receive it) until evaluation of
-the chained user commands (`expr1|expr2|...|exprN`) has completed.
-
 Job control is achieved by calling a combination of the |jobstart()|,
-|jobsend()| and |jobstop()| functions, and by listening to the |JobActivity|
-event. The best way to understand is with a complete example:
+|jobsend()| and |jobstop()| functions. Here's an example:
 >
-    let job1 = jobstart('shell1', 'bash')
-    let job2 = jobstart('shell2', 'bash', ['-c', 'for ((i = 0; i < 10; i++)); do echo hello $i!; sleep 1; done'])
-
-    function JobHandler()
-      if v:job_data[1] == 'stdout'
-        let str = 'shell '. v:job_data[0].' stdout: '.join(v:job_data[2])
-      elseif v:job_data[1] == 'stderr'
-        let str = 'shell '.v:job_data[0].' stderr: '.join(v:job_data[2])
+    function s:JobHandler(job_id, data, event)
+      if a:event == 'stdout'
+        let str = self.shell.' stdout: '.join(a:data)
+      elseif a:event == 'stderr'
+        let str = self.shell.' stderr: '.join(a:data)
       else
-        let str = 'shell '.v:job_data[0].' exited'
+        let str = self.shell.' exited'
       endif
 
       call append(line('$'), str)
     endfunction
+    let s:callbacks = {
+    \ 'on_stdout': function('s:JobHandler'),
+    \ 'on_stderr': function('s:JobHandler'),
+    \ 'on_exit': function('s:JobHandler')
+    \ }
+    let job1 = jobstart(['bash'], extend({'shell': 'shell 1'}, s:callbacks))
+    let job2 = jobstart(['bash', '-c', 'for i in {1..10}; do echo hello $i!; sleep 1; done'], extend({'shell': 'shell 2'}, s:callbacks))
 
-    au JobActivity shell* call JobHandler()
 <
 To test the above, copy it to the file ~/jobcontrol.vim and start with a clean
 nvim instance:
@@ -82,16 +73,51 @@ Here's what is happening:
 - The second shell is started with the -c argument, causing it to execute a
   command then exit. In this case, the command is a for loop that will print 0
   through 9 then exit.
-- The `JobHandler()` function is called by the `JobActivity` autocommand (notice
-  how the shell* pattern matches the names `shell1` and `shell2` passed to
-  |jobstart()|), and it takes care of displaying stdout/stderr received from
+- The `JobHandler()` function is a callback passed to |jobstart()| to handle
+  various job events. It takes care of displaying stdout/stderr received from
   the shells.
-- The v:job_data is an array set by the JobActivity event. It has the
-  following elements:
+- The arguments passed to `JobHandler()` are: 
+
   0: The job id
-  1: The kind of activity: one of "stdout", "stderr" or "exit"
-  2: When "activity" is "stdout" or "stderr", this will contain a list of
-     lines read from stdout or stderr
+  1: If the event is "stdout" or "stderr", a list with lines read from the
+     corresponding stream. For "exit", it is the status returned by the
+     program.
+  2: The event type, which is "stdout", "stderr" or "exit".
+
+The options dictionary is passed as the "self" variable to the callback
+function. Here's a more object-oriented version of the above:
+>
+    let Shell = {}
+    
+    function Shell.on_stdout(job_id, data)
+      call append(line('$'), self.get_name().' stdout: '.join(a:data))
+    endfunction
+    
+    function Shell.on_stderr(job_id, data)
+      call append(line('$'), self.get_name().' stderr: '.join(a:data))
+    endfunction
+    
+    function Shell.on_exit(job_id, data)
+      call append(line('$'), self.get_name().' exited')
+    endfunction
+    
+    function Shell.get_name()
+      return 'shell '.self.name
+    endfunction
+    
+    function Shell.new(name, ...)
+      let instance = extend(copy(g:Shell), {'name': a:name})
+      let argv = ['bash']
+      if a:0 > 0
+        let argv += ['-c', a:1]
+      endif
+      let instance.id = jobstart(argv, instance)
+      return instance
+    endfunction
+    
+    let s1 = Shell.new('1')
+    let s2 = Shell.new('2', 'for i in {1..10}; do echo hello $i!; sleep 1; done')
+
 
 To send data to the job's stdin, one can use the |jobsend()| function, like
 this:
diff --git a/runtime/doc/nvim_intro.txt b/runtime/doc/nvim_intro.txt
index d44648c575..40f65620af 100644
--- a/runtime/doc/nvim_intro.txt
+++ b/runtime/doc/nvim_intro.txt
@@ -13,13 +13,14 @@ see |help.txt|.
 For now, it is just an index with the most relevant topics/features that
 differentiate Nvim from Vim:
 
-1. Differences from Vim		|vim-differences|
-2. Msgpack-RPC			|msgpack-rpc|
-3. Job control			|job-control|
-4. Python plugins		|nvim-python|
-5. Clipboard integration	|nvim-clipboard|
-6. Remote plugins		|remote-plugin|
-7. Provider infrastructure	|nvim-provider|
+1. Differences from Vim		 |vim-differences|
+2. Msgpack-RPC			 |msgpack-rpc|
+3. Job control			 |job-control|
+4. Python plugins		 |nvim-python|
+5. Clipboard integration	 |nvim-clipboard|
+6. Remote plugins		 |remote-plugin|
+7. Provider infrastructure	 |nvim-provider|
+8. Integrated terminal emulator  |nvim-terminal-emulator|
 
 ==============================================================================
  vim:tw=78:ts=8:noet:ft=help:norl:
diff --git a/runtime/doc/nvim_terminal_emulator.txt b/runtime/doc/nvim_terminal_emulator.txt
new file mode 100644
index 0000000000..160242a7fa
--- /dev/null
+++ b/runtime/doc/nvim_terminal_emulator.txt
@@ -0,0 +1,114 @@
+*nvim_terminal_emulator.txt*    For Nvim.				{Nvim}
+
+
+		 NVIM REFERENCE MANUAL    by Thiago de Arruda
+
+
+Nvim integrated terminal emulator		     *nvim-terminal-emulator*
+
+1. Introduction			|nvim-terminal-emulator-introduction|
+2. Spawning			|nvim-terminal-emulator-spawning|
+3. Input			|nvim-terminal-emulator-input|
+4. Configuration		|nvim-terminal-emulator-configuration|
+
+==============================================================================
+1. Introduction			      *nvim-terminal-emulator-introduction*
+
+One feature that distinguishes Nvim from Vim is that it implements a mostly
+complete VT220/xterm-like terminal emulator. The terminal is presented to the
+user as a special buffer type, one that is asynchronously updated to mirror
+the virtual terminal display as data is received from the program connected
+to it. For most purposes, terminal buffers behave a lot like normal buffers
+with 'nomodifiable' set.
+
+
+The implementation is powered by libvterm[1], a powerful abstract terminal
+emulation library.
+
+[1]: http://www.leonerd.org.uk/code/libvterm/
+
+==============================================================================
+2. Spawning			      *nvim-terminal-emulator-spawning*
+
+There are 3 ways to create a terminal buffer:
+
+- By invoking the |:terminal| ex command.
+- By calling the |termopen()| function.
+- By editing a file with a name matching `term://(.{-}//(\d+:)?)?\zs.*`.
+  For example:
+>
+    :e term://bash
+    :vsp term://top
+<
+When the terminal spawns the program, the buffer will start to mirror the
+terminal display and change its name to `term://$CWD//$PID:$COMMAND`.
+Note that |:mksession| will "save" the terminal buffers by restarting all
+programs when the session is restored.
+
+==============================================================================
+3. Input			      *nvim-terminal-emulator-input*
+
+Sending input is possible by entering terminal mode, which is achieved by
+pressing any key that would enter insert mode in a normal buffer (|i| or |a|
+for example). The |:terminal| ex command will automatically enter terminal
+mode once it's spawned. While in terminal mode, Nvim will forward all keys to
+the underlying program. The only exception is the <C-\><C-n> key combo,
+which will exit back to normal mode.
+
+Terminal mode has its own namespace for mappings, which is accessed with the
+"t" prefix. It's possible to use terminal mappings to customize interaction
+with the terminal. For example, here's how to map <Esc> to exit terminal mode:
+>
+    :tnoremap <Esc> <C-\><C-n>
+<
+Navigating to other windows is only possible by exiting to normal mode, which
+can be cumbersome with <C-\><C-n> keys. Here are some mappings to improve
+the navigation experience:
+>
+    :tnoremap <A-h> <C-\><C-n><C-w>h
+    :tnoremap <A-j> <C-\><C-n><C-w>j
+    :tnoremap <A-k> <C-\><C-n><C-w>k
+    :tnoremap <A-l> <C-\><C-n><C-w>l
+    :nnoremap <A-h> <C-w>h
+    :nnoremap <A-j> <C-w>j
+    :nnoremap <A-k> <C-w>k
+    :nnoremap <A-l> <C-w>l
+<
+This allows using `Alt+{h,j,k,l}` to navigate between windows no matter if
+they are displaying a normal buffer or a terminal buffer in terminal mode.
+
+Mouse input is also fully supported, and has the following behavior:
+
+- If the program has enabled mouse events, the corresponding events will be
+  forwarded to the program.
+- If mouse events are disabled (the default), terminal focus will be lost and
+  the event will be processed as in a normal buffer.
+- If another window is clicked, terminal focus will be lost and nvim will jump
+  to the clicked window
+- If the mouse wheel is used while the mouse is positioned in another window,
+  the terminal wont lose focus and the hovered window will be scrolled.
+
+==============================================================================
+4. Configuration		     *nvim-terminal-emulator-configuration*
+
+Terminal buffers can be customized through the following global/buffer-local
+variables (set via the |TermOpen| autocmd):
+
+- `{g,b}:terminal_scrollback_buffer_size`: Scrollback buffer size, between 1
+  and 100000 inclusive. The default is 1000.
+- `{g,b}:terminal_color_$NUM`: The terminal color palette, where `$NUM` is the
+  color index, between 0 and 255 inclusive. This only affects UIs with RGB
+  capabilities; for normal terminals the color index is simply forwarded.
+- `{g,b}:terminal_focused_cursor_highlight`: Highlight group applied to the
+  cursor in a focused terminal. The default equivalent to having a group with
+  `cterm=reverse` `gui=reverse``.
+- `{g,b}:terminal_unfocused_cursor_highlight`: Highlight group applied to the
+  cursor in an unfocused terminal. The default equivalent to having a group with
+  `ctermbg=11` `guibg=#fce94f``.
+
+The configuration variables are only processed when the terminal starts, which
+is why it needs to be done with the |TermOpen| autocmd or setting global
+variables before the terminal is started.
+
+==============================================================================
+ vim:tw=78:ts=8:noet:ft=help:norl:
diff --git a/runtime/doc/options.txt b/runtime/doc/options.txt
index 0322fae991..0c964ae519 100644
--- a/runtime/doc/options.txt
+++ b/runtime/doc/options.txt
@@ -1277,6 +1277,9 @@ A jump table for the options with a short description can be found at |Q_op|.
 			or list of locations |:lwindow|
 	  help		help buffer (you are not supposed to set this
 			manually)
+	  terminal	terminal buffer, this is set automatically when a 
+			terminal is created. See |nvim-terminal-emulator| for
+			more information.
 
 	This option is used together with 'bufhidden' and 'swapfile' to
 	specify special kinds of buffers.   See |special-buffers|.
diff --git a/runtime/doc/various.txt b/runtime/doc/various.txt
index 7797b02ba8..d138fcf456 100644
--- a/runtime/doc/various.txt
+++ b/runtime/doc/various.txt
@@ -224,6 +224,16 @@ g8			Print the hex values of the bytes used in the
 						  *:sh* *:shell* *E371* *E360*
 :sh[ell]		Removed. |vim-differences| {Nvim}
 
+						  *:term* *:terminal*
+:term[inal][!] {cmd}	Spawns {command} using the current value of 'shell'
+			in a new terminal buffer.  This is equivalent to: >
+
+			      :enew | call termopen('{cmd}') | startinsert
+<
+			Like |:enew|, it will fail if the current buffer is
+			modified, but can be forced with "!".  See |termopen()|
+			and |nvim-terminal-emulator| for more information.
+
 							*:!cmd* *:!* *E34*
 :!{cmd}			Execute {cmd} with the shell. See also 'shell'.
 
diff --git a/src/nvim/eval.c b/src/nvim/eval.c
index ae00888233..3b1b034b33 100644
--- a/src/nvim/eval.c
+++ b/src/nvim/eval.c
@@ -427,7 +427,6 @@ static struct vimvar {
   {VV_NAME("oldfiles",         VAR_LIST), 0},
   {VV_NAME("windowid",         VAR_NUMBER), VV_RO},
   {VV_NAME("progpath",         VAR_STRING), VV_RO},
-  {VV_NAME("job_data",         VAR_LIST), 0},
   {VV_NAME("command_output",   VAR_STRING), 0}
 };
 
@@ -446,8 +445,11 @@ typedef struct {
   Job *job;
   Terminal *term;
   bool exited;
+  bool stdin_closed;
   int refcount;
-  char *autocmd_file;
+  ufunc_T *on_stdout, *on_stderr, *on_exit;
+  dict_T *self;
+  int *status_ptr;
 } TerminalJobData;
 
 
@@ -460,13 +462,17 @@ typedef struct {
                                    valid character */
 // Memory pool for reusing JobEvent structures
 typedef struct {
-  int id;
-  char *name, *type;
+  int job_id;
+  TerminalJobData *data;
+  ufunc_T *callback;
+  const char *type;
   list_T *received;
+  int status;
 } JobEvent;
 #define JobEventFreer(x)
 KMEMPOOL_INIT(JobEventPool, JobEvent, JobEventFreer)
 static kmempool_t(JobEventPool) *job_event_pool = NULL;
+static bool defer_job_callbacks = true;
 
 /*
  * Initialize the global and v: variables.
@@ -5933,6 +5939,41 @@ dictitem_T *dict_find(dict_T *d, char_u *key, int len)
   return HI2DI(hi);
 }
 
+// Get a function from a dictionary
+static ufunc_T *get_dict_callback(dict_T *d, char *key)
+{
+  dictitem_T *di = dict_find(d, (uint8_t *)key, -1);
+
+  if (di == NULL) {
+    return NULL;
+  }
+  
+  if (di->di_tv.v_type != VAR_FUNC && di->di_tv.v_type != VAR_STRING) {
+    EMSG(_("Argument is not a function or function name"));
+    return NULL;
+  }
+
+  uint8_t *name = di->di_tv.vval.v_string;
+  uint8_t *n = name;
+  ufunc_T *rv;
+  if (*n > '9' || *n < '0') {
+    n = trans_function_name(&n, false, TFN_INT|TFN_QUIET, NULL);
+    rv = find_func(n);
+    free(n);
+  } else {
+    // dict function, name is already translated
+    rv = find_func(n);
+  }
+
+  if (!rv) {
+    EMSG2(_("Function %s doesn't exist"), name);
+    return NULL;
+  }
+  rv->uf_refcount++;
+
+  return rv;
+}
+
 /*
  * Get a string item from a dictionary.
  * When "save" is TRUE allocate memory for it.
@@ -6495,10 +6536,12 @@ static struct fst {
   {"isdirectory",     1, 1, f_isdirectory},
   {"islocked",        1, 1, f_islocked},
   {"items",           1, 1, f_items},
+  {"jobclose",        1, 2, f_jobclose},
   {"jobresize",       3, 3, f_jobresize},
   {"jobsend",         2, 2, f_jobsend},
-  {"jobstart",        2, 4, f_jobstart},
+  {"jobstart",        1, 2, f_jobstart},
   {"jobstop",         1, 1, f_jobstop},
+  {"jobwait",         1, 2, f_jobwait},
   {"join",            1, 2, f_join},
   {"keys",            1, 1, f_keys},
   {"last_buffer_nr",  0, 0, f_last_buffer_nr},  /* obsolete */
@@ -6917,24 +6960,9 @@ call_func (
         else if ((fp->uf_flags & FC_DICT) && selfdict == NULL)
           error = ERROR_DICT;
         else {
-          /*
-           * Call the user function.
-           * Save and restore search patterns, script variables and
-           * redo buffer.
-           */
-          save_search_patterns();
-          saveRedobuff();
-          ++fp->uf_calls;
-          call_user_func(fp, argcount, argvars, rettv,
-              firstline, lastline,
+          // Call the user function.
+          call_user_func(fp, argcount, argvars, rettv, firstline, lastline,
               (fp->uf_flags & FC_DICT) ? selfdict : NULL);
-          if (--fp->uf_calls <= 0 && isdigit(*fp->uf_name)
-              && fp->uf_refcount <= 0)
-            /* Function was unreferenced while being used, free it
-             * now. */
-            func_free(fp);
-          restoreRedobuff();
-          restore_search_patterns();
           error = ERROR_NONE;
         }
       }
@@ -10632,6 +10660,48 @@ static void f_items(typval_T *argvars, typval_T *rettv)
   dict_list(argvars, rettv, 2);
 }
 
+// "jobclose(id[, stream])" function
+static void f_jobclose(typval_T *argvars, typval_T *rettv)
+{
+  rettv->v_type = VAR_NUMBER;
+  rettv->vval.v_number = 0;
+
+  if (check_restricted() || check_secure()) {
+    return;
+  }
+
+  if (argvars[0].v_type != VAR_NUMBER || (argvars[1].v_type != VAR_STRING
+        && argvars[1].v_type != VAR_UNKNOWN)) {
+    EMSG(_(e_invarg));
+    return;
+  }
+
+  Job *job = job_find(argvars[0].vval.v_number);
+
+  if (!is_user_job(job)) {
+    // Invalid job id
+    EMSG(_(e_invjob));
+    return;
+  }
+
+  if (argvars[1].v_type == VAR_STRING) {
+    char *stream = (char *)argvars[1].vval.v_string;
+    if (!strcmp(stream, "stdin")) {
+      job_close_in(job);
+      ((TerminalJobData *)job_data(job))->stdin_closed = true;
+    } else if (!strcmp(stream, "stdout")) {
+      job_close_out(job);
+    } else if (!strcmp(stream, "stderr")) {
+      job_close_err(job);
+    } else {
+      EMSG2(_("Invalid job stream \"%s\""), stream);
+    }
+  } else {
+    ((TerminalJobData *)job_data(job))->stdin_closed = true;
+    job_close_streams(job);
+  }
+}
+
 // "jobsend()" function
 static void f_jobsend(typval_T *argvars, typval_T *rettv)
 {
@@ -10651,12 +10721,17 @@ static void f_jobsend(typval_T *argvars, typval_T *rettv)
 
   Job *job = job_find(argvars[0].vval.v_number);
 
-  if (!job) {
+  if (!is_user_job(job)) {
     // Invalid job id
     EMSG(_(e_invjob));
     return;
   }
 
+  if (((TerminalJobData *)job_data(job))->stdin_closed) {
+    EMSG(_("Can't send data to the job: stdin is closed"));
+    return;
+  }
+
   ssize_t input_len;
   char *input = (char *) save_tv_as_string(&argvars[1], &input_len, false);
   if (!input) {
@@ -10669,7 +10744,7 @@ static void f_jobsend(typval_T *argvars, typval_T *rettv)
   rettv->vval.v_number = job_write(job, buf);
 }
 
-// "jobresize()" function
+// "jobresize(job, width, height)" function
 static void f_jobresize(typval_T *argvars, typval_T *rettv)
 {
   rettv->v_type = VAR_NUMBER;
@@ -10688,7 +10763,7 @@ static void f_jobresize(typval_T *argvars, typval_T *rettv)
 
   Job *job = job_find(argvars[0].vval.v_number);
 
-  if (!job) {
+  if (!is_user_job(job)) {
     // Probably an invalid job id
     EMSG(_(e_invjob));
     return;
@@ -10705,11 +10780,6 @@ static void f_jobresize(typval_T *argvars, typval_T *rettv)
 // "jobstart()" function
 static void f_jobstart(typval_T *argvars, typval_T *rettv)
 {
-  list_T *args = NULL;
-  listitem_T *arg;
-  int i, argvl, argsl;
-  char **argv = NULL;
-
   rettv->v_type = VAR_NUMBER;
   rettv->vval.v_number = 0;
 
@@ -10717,55 +10787,60 @@ static void f_jobstart(typval_T *argvars, typval_T *rettv)
     return;
   }
 
-  if (argvars[0].v_type != VAR_STRING
-      || argvars[1].v_type != VAR_STRING
-      || (argvars[2].v_type != VAR_LIST && argvars[2].v_type != VAR_UNKNOWN)) {
+  if (argvars[0].v_type != VAR_LIST
+      || (argvars[1].v_type != VAR_DICT && argvars[1].v_type != VAR_UNKNOWN)) {
     // Wrong argument types
     EMSG(_(e_invarg));
     return;
   }
 
-  argsl = 0;
-  if (argvars[2].v_type == VAR_LIST) {
-    args = argvars[2].vval.v_list;
-    argsl = args->lv_len;
-    // Assert that all list items are strings
-    for (arg = args->lv_first; arg != NULL; arg = arg->li_next) {
-      if (arg->li_tv.v_type != VAR_STRING) {
-        EMSG(_(e_invarg));
-        return;
-      }
+  list_T *args = argvars[0].vval.v_list;
+  // Assert that all list items are strings
+  for (listitem_T *arg = args->lv_first; arg != NULL; arg = arg->li_next) {
+    if (arg->li_tv.v_type != VAR_STRING) {
+      EMSG(_(e_invarg));
+      return;
     }
   }
 
-  if (!os_can_exe(get_tv_string(&argvars[1]), NULL)) {
-    // String is not executable
-    EMSG2(e_jobexe, get_tv_string(&argvars[1]));
+  int argc = args->lv_len;
+  if (!argc) {
+    EMSG(_("Argument vector must have at least one item"));
     return;
   }
 
-  // Allocate extra memory for the argument vector and the NULL pointer
-  argvl = argsl + 2;
-  argv = xmalloc(sizeof(char_u *) * argvl);
-
-  // Copy program name
-  argv[0] = xstrdup((char *)get_tv_string(&argvars[1]));
+  if (!os_can_exe(args->lv_first->li_tv.vval.v_string, NULL)) {
+    // String is not executable
+    EMSG2(e_jobexe, args->lv_first->li_tv.vval.v_string);
+    return;
+  }
 
-  i = 1;
-  // Copy arguments to the vector
-  if (argsl > 0) {
-    for (arg = args->lv_first; arg != NULL; arg = arg->li_next) {
-      argv[i++] = xstrdup((char *)get_tv_string(&arg->li_tv));
+  dict_T *job_opts = NULL;
+  ufunc_T *on_stdout = NULL, *on_stderr = NULL, *on_exit = NULL;
+  if (argvars[1].v_type == VAR_DICT) {
+    job_opts = argvars[1].vval.v_dict;
+    common_job_callbacks(job_opts, &on_stdout, &on_stderr, &on_exit);
+    if (did_emsg) {
+      return;
     }
   }
 
-  // The last item of argv must be NULL
-  argv[i] = NULL;
-  JobOptions opts = common_job_options(argv, (char *)argvars[0].vval.v_string);
+  // Build the argument vector
+  int i = 0;
+  char **argv = xcalloc(argc + 1, sizeof(char *));
+  for (listitem_T *arg = args->lv_first; arg != NULL; arg = arg->li_next) {
+    argv[i++] = xstrdup((char *)arg->li_tv.vval.v_string);
+  }
+
+  JobOptions opts = common_job_options(argv, on_stdout, on_stderr, on_exit,
+      job_opts);
+
+  if (!job_opts) {
+    goto start;
+  }
 
-  if (args && argvars[3].v_type == VAR_DICT) {
-    dict_T *job_opts = argvars[3].vval.v_dict;
-    opts.pty = true;
+  opts.pty = get_dict_number(job_opts, (uint8_t *)"pty");
+  if (opts.pty) {
     uint16_t width = get_dict_number(job_opts, (uint8_t *)"width");
     if (width > 0) {
       opts.width = width;
@@ -10780,6 +10855,13 @@ static void f_jobstart(typval_T *argvars, typval_T *rettv)
     }
   }
 
+start:
+  if (!on_stdout) {
+    opts.stdout_cb = NULL;
+  }
+  if (!on_stderr) {
+    opts.stderr_cb = NULL;
+  }
   common_job_start(opts, rettv);
 }
 
@@ -10801,8 +10883,7 @@ static void f_jobstop(typval_T *argvars, typval_T *rettv)
 
   Job *job = job_find(argvars[0].vval.v_number);
 
-  if (!job) {
-    // Probably an invalid job id
+  if (!is_user_job(job)) {
     EMSG(_(e_invjob));
     return;
   }
@@ -10811,6 +10892,105 @@ static void f_jobstop(typval_T *argvars, typval_T *rettv)
   rettv->vval.v_number = 1;
 }
 
+// "jobwait(ids[, timeout])" function
+static void f_jobwait(typval_T *argvars, typval_T *rettv)
+{
+  rettv->v_type = VAR_NUMBER;
+  rettv->vval.v_number = 0;
+
+  if (check_restricted() || check_secure()) {
+    return;
+  }
+
+  if (argvars[0].v_type != VAR_LIST || (argvars[1].v_type != VAR_NUMBER
+        && argvars[1].v_type != VAR_UNKNOWN)) {
+    EMSG(_(e_invarg));
+    return;
+  }
+
+  list_T *args = argvars[0].vval.v_list;
+  list_T *rv = list_alloc();
+
+  // must temporarily disable job event deferring so the callbacks are
+  // processed while waiting.
+  defer_job_callbacks = false;
+  // For each item in the input list append an integer to the output list. -3
+  // is used to represent an invalid job id, -2 is for a interrupted job and
+  // -1 for jobs that were skipped or timed out.
+  for (listitem_T *arg = args->lv_first; arg != NULL; arg = arg->li_next) {
+    Job *job = NULL;
+    if (arg->li_tv.v_type != VAR_NUMBER
+        || !(job = job_find(arg->li_tv.vval.v_number))
+        || !is_user_job(job)) {
+      list_append_number(rv, -3);
+    } else {
+      TerminalJobData *data = job_data(job);
+      // append the list item and set the status pointer so we'll collect the
+      // status code when the job exits
+      list_append_number(rv, -1);
+      data->status_ptr = &rv->lv_last->li_tv.vval.v_number;
+    }
+  }
+
+  int remaining = -1;
+  uint64_t before = 0;
+  if (argvars[1].v_type == VAR_NUMBER && argvars[1].vval.v_number >= 0) {
+    remaining = argvars[1].vval.v_number;
+    before = os_hrtime();
+  }
+
+  for (listitem_T *arg = args->lv_first; arg != NULL; arg = arg->li_next) {
+    Job *job = NULL;
+    if (remaining == 0) {
+      // timed out
+      break;
+    }
+    if (arg->li_tv.v_type != VAR_NUMBER
+        || !(job = job_find(arg->li_tv.vval.v_number))
+        || !is_user_job(job)) {
+      continue;
+    }
+    TerminalJobData *data = job_data(job);
+    int status = job_wait(job, remaining);
+    if (status < 0) {
+      // interrupted or timed out, skip remaining jobs.
+      if (status == -2) {
+        // set the status so the user can distinguish between interrupted and
+        // skipped/timeout jobs.
+        *data->status_ptr = -2;
+      }
+      break;
+    }
+    if (remaining > 0) {
+      uint64_t now = os_hrtime();
+      remaining -= (int) ((now - before) / 1000000);
+      before = now;
+      if (remaining <= 0) {
+        break;
+      }
+    }
+  }
+
+  for (listitem_T *arg = args->lv_first; arg != NULL; arg = arg->li_next) {
+    Job *job = NULL;
+    if (arg->li_tv.v_type != VAR_NUMBER
+        || !(job = job_find(arg->li_tv.vval.v_number))
+        || !is_user_job(job)) {
+      continue;
+    }
+    TerminalJobData *data = job_data(job);
+    // remove the status pointer because the list may be freed before the
+    // job exits
+    data->status_ptr = NULL;
+  }
+  // restore defer flag
+  defer_job_callbacks = true;
+
+  rv->lv_refcount++;
+  rettv->v_type = VAR_LIST;
+  rettv->vval.v_list = rv;
+}
+
 /*
  * "join()" function
  */
@@ -14881,15 +15061,25 @@ static void f_termopen(typval_T *argvars, typval_T *rettv)
   }
 
   if (argvars[0].v_type != VAR_STRING
-      || (argvars[1].v_type != VAR_STRING
-        && argvars[1].v_type != VAR_UNKNOWN)) {
+      || (argvars[1].v_type != VAR_DICT && argvars[1].v_type != VAR_UNKNOWN)) {
     // Wrong argument types
     EMSG(_(e_invarg));
     return;
   }
 
+  ufunc_T *on_stdout = NULL, *on_stderr = NULL, *on_exit = NULL;
+  dict_T *job_opts = NULL;
+  if (argvars[1].v_type == VAR_DICT) {
+    job_opts = argvars[1].vval.v_dict;
+    common_job_callbacks(job_opts, &on_stdout, &on_stderr, &on_exit);
+    if (did_emsg) {
+      return;
+    }
+  }
+
   char **argv = shell_build_argv((char *)argvars[0].vval.v_string, NULL);
-  JobOptions opts = common_job_options(argv, NULL);
+  JobOptions opts = common_job_options(argv, on_stdout, on_stderr, on_exit,
+      job_opts);
   opts.pty = true;
   opts.width = curwin->w_width;
   opts.height = curwin->w_height;
@@ -14921,7 +15111,6 @@ static void f_termopen(typval_T *argvars, typval_T *rettv)
   // the 'swapfile' option to ensure no swap file will be created
   curbuf->b_p_swf = false;
   (void)setfname(curbuf, (uint8_t *)buf, NULL, true);
-  data->autocmd_file = xstrdup(buf);
   // Save the job id and pid in b:terminal_job_{id,pid}
   Error err;
   dict_set_value(curbuf->b_vars, cstr_as_string("terminal_job_id"),
@@ -17836,7 +18025,6 @@ void ex_function(exarg_T *eap)
       fudi.fd_di->di_tv.v_type = VAR_FUNC;
       fudi.fd_di->di_tv.v_lock = 0;
       fudi.fd_di->di_tv.vval.v_string = vim_strsave(name);
-      fp->uf_refcount = 1;
 
       /* behave like "dict" was used */
       flags |= FC_DICT;
@@ -17846,6 +18034,7 @@ void ex_function(exarg_T *eap)
     STRCPY(fp->uf_name, name);
     hash_add(&func_hashtab, UF2HIKEY(fp));
   }
+  fp->uf_refcount = 1;
   fp->uf_args = newargs;
   fp->uf_lines = newlines;
   fp->uf_tml_count = NULL;
@@ -18519,6 +18708,11 @@ void ex_delfunction(exarg_T *eap)
       EMSG2(_("E131: Cannot delete function %s: It is in use"), eap->arg);
       return;
     }
+    if (fp->uf_refcount > 1) {
+      EMSG2(_("Cannot delete function %s: It is being used internally"),
+          eap->arg);
+      return;
+    }
 
     if (fudi.fd_dict != NULL) {
       /* Delete the dict item that refers to the function, it will
@@ -18563,13 +18757,21 @@ void func_unref(char_u *name)
 
   if (name != NULL && isdigit(*name)) {
     fp = find_func(name);
-    if (fp == NULL)
+    if (fp == NULL) {
       EMSG2(_(e_intern2), "func_unref()");
-    else if (--fp->uf_refcount <= 0) {
-      /* Only delete it when it's not being used.  Otherwise it's done
-       * when "uf_calls" becomes zero. */
-      if (fp->uf_calls == 0)
-        func_free(fp);
+    } else {
+      user_func_unref(fp);
+    }
+  }
+}
+
+static void user_func_unref(ufunc_T *fp)
+{
+  if (--fp->uf_refcount <= 0) {
+    // Only delete it when it's not being used.  Otherwise it's done
+    // when "uf_calls" becomes zero.
+    if (fp->uf_calls == 0) {
+      func_free(fp);
     }
   }
 }
@@ -18626,9 +18828,13 @@ call_user_func (
     return;
   }
   ++depth;
-
-  line_breakcheck();            /* check for CTRL-C hit */
-
+  // Save search patterns and redo buffer.
+  save_search_patterns();
+  saveRedobuff();
+  ++fp->uf_calls;
+  // check for CTRL-C hit
+  line_breakcheck();
+  // prepare the funccall_T structure
   fc = xmalloc(sizeof(funccall_T));
   fc->caller = current_funccal;
   current_funccal = fc;
@@ -18924,6 +19130,14 @@ call_user_func (
     for (li = fc->l_varlist.lv_first; li != NULL; li = li->li_next)
       copy_tv(&li->li_tv, &li->li_tv);
   }
+
+  if (--fp->uf_calls <= 0 && isdigit(*fp->uf_name) && fp->uf_refcount <= 0) {
+    // Function was unreferenced while being used, free it now.
+    func_free(fp);
+  }
+  // restore search patterns and redo buffer
+  restoreRedobuff();
+  restore_search_patterns();
 }
 
 /*
@@ -19814,12 +20028,14 @@ char_u *do_string_sub(char_u *str, char_u *pat, char_u *sub, char_u *flags)
   return ret;
 }
 
-static inline JobOptions common_job_options(char **argv, char *autocmd_file)
+static inline JobOptions common_job_options(char **argv, ufunc_T *on_stdout,
+    ufunc_T *on_stderr, ufunc_T *on_exit, dict_T *self)
 {
   TerminalJobData *data = xcalloc(1, sizeof(TerminalJobData));
-  if (autocmd_file) {
-    data->autocmd_file = xstrdup(autocmd_file);
-  }
+  data->on_stdout = on_stdout;
+  data->on_stderr = on_stderr;
+  data->on_exit = on_exit;
+  data->self = self;
   JobOptions opts = JOB_OPTIONS_INIT;
   opts.argv = argv;
   opts.data = data;
@@ -19829,6 +20045,28 @@ static inline JobOptions common_job_options(char **argv, char *autocmd_file)
   return opts;
 }
 
+static inline void common_job_callbacks(dict_T *vopts, ufunc_T **on_stdout,
+    ufunc_T **on_stderr, ufunc_T **on_exit)
+{
+  *on_stdout = get_dict_callback(vopts, "on_stdout");
+  *on_stderr = get_dict_callback(vopts, "on_stderr");
+  *on_exit = get_dict_callback(vopts, "on_exit");
+  if (did_emsg) {
+    if (*on_stdout) {
+      user_func_unref(*on_stdout);
+    }
+    if (*on_stderr) {
+      user_func_unref(*on_stderr);
+    }
+    if (*on_exit) {
+      user_func_unref(*on_exit);
+    }
+    return;
+  }
+
+  vopts->dv_refcount++;
+}
+
 static inline Job *common_job_start(JobOptions opts, typval_T *rettv)
 {
   TerminalJobData *data = opts.data;
@@ -19839,8 +20077,7 @@ static inline Job *common_job_start(JobOptions opts, typval_T *rettv)
     if (rettv->vval.v_number == 0) {
       EMSG(_(e_jobtblfull));
       free(opts.term_name);
-      free(data->autocmd_file);
-      free(data);
+      free_term_job_data(data);
     } else {
       EMSG(_(e_jobexe));
     }
@@ -19850,10 +20087,33 @@ static inline Job *common_job_start(JobOptions opts, typval_T *rettv)
   return job;
 }
 
-// JobActivity autocommands will execute vimscript code, so it must be executed
-// on Nvim main loop
-static inline void push_job_event(Job *job, char *type, char *data,
-    size_t count)
+static inline void free_term_job_data(TerminalJobData *data) {
+  if (data->on_stdout) {
+    user_func_unref(data->on_stdout);
+  }
+  if (data->on_stderr) {
+    user_func_unref(data->on_stderr);
+  }
+  if (data->on_exit) {
+    user_func_unref(data->on_exit);
+  }
+  dict_unref(data->self);
+  free(data);
+}
+
+static inline bool is_user_job(Job *job)
+{
+  if (!job) {
+    return false;
+  }
+
+  JobOptions *opts = job_opts(job);
+  return opts->exit_cb == on_job_exit;
+}
+
+// vimscript job callbacks must be executed on Nvim main loop
+static inline void push_job_event(Job *job, ufunc_T *callback,
+    const char *type, char *data, size_t count, int status)
 {
   JobEvent *event_data = kmp_alloc(JobEventPool, job_event_pool);
   event_data->received = NULL;
@@ -19880,28 +20140,33 @@ static inline void push_job_event(Job *job, char *type, char *data,
       off++;
     }
     list_append_string(event_data->received, (uint8_t *)ptr, off);
+  } else {
+    event_data->status = status;
   }
-  TerminalJobData *d = job_data(job);
-  event_data->id = job_id(job);
-  event_data->name = d->autocmd_file;
+  event_data->job_id = job_id(job);
+  event_data->data = job_data(job);
+  event_data->callback = callback;
   event_data->type = type;
   event_push((Event) {
     .handler = on_job_event,
     .data = event_data
-  }, true);
+  }, defer_job_callbacks);
 }
 
-static void on_job_stdout(RStream *rstream, void *data, bool eof)
+static void on_job_stdout(RStream *rstream, void *job, bool eof)
 {
-  on_job_output(rstream, data, eof, "stdout");
+  TerminalJobData *data = job_data(job);
+  on_job_output(rstream, job, eof, data->on_stdout, "stdout");
 }
 
-static void on_job_stderr(RStream *rstream, void *data, bool eof)
+static void on_job_stderr(RStream *rstream, void *job, bool eof)
 {
-  on_job_output(rstream, data, eof, "stderr");
+  TerminalJobData *data = job_data(job);
+  on_job_output(rstream, job, eof, data->on_stderr, "stderr");
 }
 
-static void on_job_output(RStream *rstream, Job *job, bool eof, char *type)
+static void on_job_output(RStream *rstream, Job *job, bool eof,
+    ufunc_T *callback, const char *type)
 {
   if (eof) {
     return;
@@ -19917,21 +20182,28 @@ static void on_job_output(RStream *rstream, Job *job, bool eof, char *type)
     terminal_receive(data->term, ptr, len);
   }
 
-  push_job_event(job, type, ptr, len);
+  if (callback) {
+    push_job_event(job, callback, type, ptr, len, 0);
+  }
+
   rbuffer_consumed(rstream_buffer(rstream), len);
 }
 
-static void on_job_exit(Job *job, void *d)
+static void on_job_exit(Job *job, int status, void *d)
 {
   TerminalJobData *data = d;
-  push_job_event(job, "exit", NULL, 0);
 
   if (data->term && !data->exited) {
     data->exited = true;
     terminal_close(data->term,
         _("\r\n[Program exited, press any key to close]"));
   }
-  term_job_data_decref(data);
+
+  if (data->status_ptr) {
+    *data->status_ptr = status;
+  }
+
+  push_job_event(job, data->on_exit, "exit", NULL, 0, status);
 }
 
 static void term_write(char *buf, size_t size, void *data)
@@ -19960,42 +20232,57 @@ static void term_close(void *d)
 static void term_job_data_decref(TerminalJobData *data)
 {
   if (!(--data->refcount)) {
-    free(data);
+    free_term_job_data(data);
   }
 }
 
 static void on_job_event(Event event)
 {
-  JobEvent *data = event.data;
-  apply_job_autocmds(data->id, data->name, data->type, data->received);
-  kmp_free(JobEventPool, job_event_pool, data);
-}
+  JobEvent *ev = event.data;
 
-static void apply_job_autocmds(int id, char *name, char *type,
-                               list_T *received)
-{
-  // Create the list which will be set to v:job_data
-  list_T *list = list_alloc();
-  list_append_number(list, id);
-  list_append_string(list, (uint8_t *)type, -1);
+  if (!ev->callback) {
+    goto end;
+  }
 
-  if (received) {
-    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 = received;
-    str_slot->li_tv.vval.v_list->lv_refcount++;
-    list_append(list, str_slot);
+  typval_T argv[3];
+  int argc = ev->callback->uf_args.ga_len;
+
+  if (argc > 0) {
+    argv[0].v_type = VAR_NUMBER;
+    argv[0].v_lock = 0;
+    argv[0].vval.v_number = ev->job_id;
   }
 
-  // Update v:job_data for the autocommands
-  set_vim_var_list(VV_JOB_DATA, list);
-  // Call JobActivity autocommands
-  apply_autocmds(EVENT_JOBACTIVITY, (uint8_t *)name, NULL, TRUE, NULL);
+  if (argc > 1) {
+    if (ev->received) {
+      argv[1].v_type = VAR_LIST;
+      argv[1].v_lock = 0;
+      argv[1].vval.v_list = ev->received;
+      argv[1].vval.v_list->lv_refcount++;
+    } else {
+      argv[1].v_type = VAR_NUMBER;
+      argv[1].v_lock = 0;
+      argv[1].vval.v_number = ev->status;
+    }
+  }
 
-  if (!received) {
-    // This must be the exit event. Free the name.
-    free(name);
+  if (argc > 2) {
+    argv[2].v_type = VAR_STRING;
+    argv[2].v_lock = 0;
+    argv[2].vval.v_string = (uint8_t *)ev->type;
+  }
+
+  typval_T rettv;
+  init_tv(&rettv);
+  call_user_func(ev->callback, argc, argv, &rettv, curwin->w_cursor.lnum,
+      curwin->w_cursor.lnum, ev->data->self);
+  clear_tv(&rettv);
+
+end:
+  kmp_free(JobEventPool, job_event_pool, ev);
+  if (!ev->received) {
+    // exit event, safe to free job data now
+    term_job_data_decref(ev->data);
   }
 }
 
diff --git a/src/nvim/eval.h b/src/nvim/eval.h
index e96106dfb3..61a65df7c6 100644
--- a/src/nvim/eval.h
+++ b/src/nvim/eval.h
@@ -63,7 +63,6 @@ enum {
     VV_OLDFILES,
     VV_WINDOWID,
     VV_PROGPATH,
-    VV_JOB_DATA,
     VV_COMMAND_OUTPUT,
     VV_LEN, /* number of v: vars */
 };
diff --git a/src/nvim/fileio.c b/src/nvim/fileio.c
index 29bcaec84a..80f843048a 100644
--- a/src/nvim/fileio.c
+++ b/src/nvim/fileio.c
@@ -5187,7 +5187,6 @@ static struct event_name {
   {"InsertEnter",     EVENT_INSERTENTER},
   {"InsertLeave",     EVENT_INSERTLEAVE},
   {"InsertCharPre",   EVENT_INSERTCHARPRE},
-  {"JobActivity",     EVENT_JOBACTIVITY},
   {"MenuPopup",       EVENT_MENUPOPUP},
   {"QuickFixCmdPost", EVENT_QUICKFIXCMDPOST},
   {"QuickFixCmdPre",  EVENT_QUICKFIXCMDPRE},
@@ -6595,7 +6594,6 @@ apply_autocmds_group (
         || event == EVENT_QUICKFIXCMDPRE
         || event == EVENT_COLORSCHEME
         || event == EVENT_QUICKFIXCMDPOST
-        || event == EVENT_JOBACTIVITY
         || event == EVENT_TABCLOSED)
       fname = vim_strsave(fname);
     else
diff --git a/src/nvim/fileio.h b/src/nvim/fileio.h
index 833a4fade6..3b37d359d7 100644
--- a/src/nvim/fileio.h
+++ b/src/nvim/fileio.h
@@ -63,7 +63,6 @@ typedef enum auto_event {
   EVENT_INSERTCHANGE,           /* when changing Insert/Replace mode */
   EVENT_INSERTENTER,            /* when entering Insert mode */
   EVENT_INSERTLEAVE,            /* when leaving Insert mode */
-  EVENT_JOBACTIVITY,            /* when job sent some data */
   EVENT_MENUPOPUP,              /* just before popup menu is displayed */
   EVENT_QUICKFIXCMDPOST,        /* after :make, :grep etc. */
   EVENT_QUICKFIXCMDPRE,         /* before :make, :grep etc. */
diff --git a/src/nvim/main.c b/src/nvim/main.c
index 47bb2bc515..55935564b8 100644
--- a/src/nvim/main.c
+++ b/src/nvim/main.c
@@ -293,8 +293,8 @@ int main(int argc, char **argv)
       "matchstr(expand(\"<amatch>\"), "
       "'\\c\\mterm://\\%(.\\{-}//\\%(\\d\\+:\\)\\?\\)\\?\\zs.*'), "
       // capture the working directory
-      "get(matchlist(expand(\"<amatch>\"), "
-      "'\\c\\mterm://\\(.\\{-}\\)//'), 1, ''))");
+      "{'cwd': get(matchlist(expand(\"<amatch>\"), "
+      "'\\c\\mterm://\\(.\\{-}\\)//'), 1, '')})");
 
   /* Execute --cmd arguments. */
   exe_pre_commands(&params);
diff --git a/src/nvim/msgpack_rpc/channel.c b/src/nvim/msgpack_rpc/channel.c
index e8dab05ba0..b1f0798528 100644
--- a/src/nvim/msgpack_rpc/channel.c
+++ b/src/nvim/msgpack_rpc/channel.c
@@ -347,7 +347,7 @@ static void job_err(RStream *rstream, void *data, bool eof)
   }
 }
 
-static void job_exit(Job *job, void *data)
+static void job_exit(Job *job, int status, void *data)
 {
   decref(data);
 }
diff --git a/src/nvim/os/job.c b/src/nvim/os/job.c
index ccd7891601..8fe44c7a53 100644
--- a/src/nvim/os/job.c
+++ b/src/nvim/os/job.c
@@ -24,7 +24,6 @@
 // before we send SIGNAL to it
 #define TERM_TIMEOUT 1000000000
 #define KILL_TIMEOUT (TERM_TIMEOUT * 2)
-#define MAX_RUNNING_JOBS 100
 #define JOB_BUFFER_SIZE 0xFFFF
 
 #define close_job_stream(job, stream, type)                                \
@@ -234,11 +233,12 @@ void job_stop(Job *job)
 /// @return returns the status code of the exited job. -1 if the job is
 ///         still running and the `timeout` has expired. Note that this is
 ///         indistinguishable from the process returning -1 by itself. Which
-///         is possible on some OS.
+///         is possible on some OS. Returns -2 if the job was interrupted.
 int job_wait(Job *job, int ms) FUNC_ATTR_NONNULL_ALL
 {
   // The default status is -1, which represents a timeout
   int status = -1;
+  bool interrupted = false;
 
   // Increase refcount to stop the job from being freed before we have a
   // chance to get the status.
@@ -251,6 +251,7 @@ int job_wait(Job *job, int ms) FUNC_ATTR_NONNULL_ALL
   // we'll assume that a user frantically hitting interrupt doesn't like
   // the current job. Signal that it has to be killed.
   if (got_int) {
+    interrupted = true;
     got_int = false;
     job_stop(job);
     if (ms == -1) {
@@ -265,7 +266,7 @@ int job_wait(Job *job, int ms) FUNC_ATTR_NONNULL_ALL
   if (job->refcount == 1) {
     // Job exited, collect status and manually invoke close_cb to free the job
     // resources
-    status = job->status;
+    status = interrupted ? -2 : job->status;
     job_close_streams(job);
     job_decref(job);
   } else {
@@ -289,6 +290,18 @@ void job_close_in(Job *job) FUNC_ATTR_NONNULL_ALL
   close_job_in(job);
 }
 
+// Close the job stdout stream.
+void job_close_out(Job *job) FUNC_ATTR_NONNULL_ALL
+{
+  close_job_out(job);
+}
+
+// Close the job stderr stream.
+void job_close_err(Job *job) FUNC_ATTR_NONNULL_ALL
+{
+  close_job_out(job);
+}
+
 /// All writes that complete after calling this function will be reported
 /// to `cb`.
 ///
@@ -357,6 +370,11 @@ void job_close_streams(Job *job)
   close_job_err(job);
 }
 
+JobOptions *job_opts(Job *job)
+{
+  return &job->opts;
+}
+
 /// Iterates the table, sending SIGTERM to stopped jobs and SIGKILL to those
 /// that didn't die from SIGTERM after a while(exit_timeout is 0).
 static void job_stop_timer_cb(uv_timer_t *handle)
diff --git a/src/nvim/os/job_defs.h b/src/nvim/os/job_defs.h
index ac9a37b366..7fee900ac0 100644
--- a/src/nvim/os/job_defs.h
+++ b/src/nvim/os/job_defs.h
@@ -5,13 +5,14 @@
 #include "nvim/os/rstream_defs.h"
 #include "nvim/os/wstream_defs.h"
 
+#define MAX_RUNNING_JOBS 100
 typedef struct job Job;
 
 /// Function called when the job reads data
 ///
 /// @param id The job id
 /// @param data Some data associated with the job by the caller
-typedef void (*job_exit_cb)(Job *job, void *data);
+typedef void (*job_exit_cb)(Job *job, int status, void *data);
 
 // Job startup options
 // job_exit_cb Callback that will be invoked when the job exits
diff --git a/src/nvim/os/job_private.h b/src/nvim/os/job_private.h
index b1d5e13feb..af13d2e636 100644
--- a/src/nvim/os/job_private.h
+++ b/src/nvim/os/job_private.h
@@ -88,7 +88,7 @@ static inline void job_exit_callback(Job *job)
 
   if (job->opts.exit_cb) {
     // Invoke the exit callback
-    job->opts.exit_cb(job, job->opts.data);
+    job->opts.exit_cb(job, job->status, job->opts.data);
   }
 
   if (stop_requests && !--stop_requests) {
diff --git a/test/functional/job/job_spec.lua b/test/functional/job/job_spec.lua
index 8981c49744..c1c559eb34 100644
--- a/test/functional/job/job_spec.lua
+++ b/test/functional/job/job_spec.lua
@@ -1,10 +1,11 @@
 
 local helpers = require('test.functional.helpers')
-local clear, nvim, eq, neq, ok, expect, eval, next_message, run, stop, session
+local clear, nvim, eq, neq, ok, expect, eval, next_msg, run, stop, session
   = helpers.clear, helpers.nvim, helpers.eq, helpers.neq, helpers.ok,
   helpers.expect, helpers.eval, helpers.next_message, helpers.run,
   helpers.stop, helpers.session
-local nvim_dir, insert = helpers.nvim_dir, helpers.insert
+local nvim_dir, insert, feed = helpers.nvim_dir, helpers.insert, helpers.feed
+local source, execute, wait = helpers.source, helpers.execute, helpers.wait
 
 
 describe('jobs', function()
@@ -13,46 +14,44 @@ describe('jobs', function()
   before_each(function()
     clear()
     channel = nvim('get_api_info')[1]
+    nvim('set_var', 'channel', channel)
+    source([[
+    function! s:OnEvent(id, data, event)
+      let userdata = get(self, 'user')
+      call rpcnotify(g:channel, a:event, userdata, a:data)
+    endfunction
+    let g:job_opts = {
+    \ 'on_stdout': function('s:OnEvent'),
+    \ 'on_stderr': function('s:OnEvent'),
+    \ 'on_exit': function('s:OnEvent'),
+    \ 'user': 0
+    \ }
+    ]])
   end)
 
-  -- Creates the string to make an autocmd to notify us.
-  local notify_str = function(expr1, expr2)
-    local str = "au! JobActivity xxx call rpcnotify("..channel..", "..expr1
-    if expr2 ~= nil then
-      str = str..", "..expr2
-    end
-    return str..")"
-  end
-
-  local notify_job = function()
-    return "au! JobActivity xxx call rpcnotify("..channel..", 'j', v:job_data)"
-  end
-
   it('returns 0 when it fails to start', function()
-    local status, rv = pcall(eval, "jobstart('', '')")
+    local status, rv = pcall(eval, "jobstart([])")
     eq(false, status)
     ok(rv ~= nil)
   end)
 
-  it('calls JobActivity when the job writes and exits', function()
-    nvim('command', notify_str('v:job_data[1]'))
-    nvim('command', "call jobstart('xxx', 'echo')")
-    eq({'notification', 'stdout', {}}, next_message())
-    eq({'notification', 'exit', {}}, next_message())
+  it('invokes callbacks when the job writes and exits', function()
+    nvim('command', "call jobstart(['echo'], g:job_opts)")
+    eq({'notification', 'stdout', {0, {'', ''}}}, next_msg())
+    eq({'notification', 'exit', {0, 0}}, next_msg())
   end)
 
   it('allows interactive commands', function()
-    nvim('command', notify_str('v:job_data[1]', 'get(v:job_data, 2)'))
-    nvim('command', "let j = jobstart('xxx', 'cat', ['-'])")
+    nvim('command', "let j = jobstart(['cat', '-'], g:job_opts)")
     neq(0, eval('j'))
     nvim('command', 'call jobsend(j, "abc\\n")')
-    eq({'notification', 'stdout', {{'abc', ''}}}, next_message())
+    eq({'notification', 'stdout', {0, {'abc', ''}}}, next_msg())
     nvim('command', 'call jobsend(j, "123\\nxyz\\n")')
-    eq({'notification', 'stdout', {{'123', 'xyz', ''}}}, next_message())
+    eq({'notification', 'stdout', {0, {'123', 'xyz', ''}}}, next_msg())
     nvim('command', 'call jobsend(j, [123, "xyz", ""])')
-    eq({'notification', 'stdout', {{'123', 'xyz', ''}}}, next_message())
+    eq({'notification', 'stdout', {0, {'123', 'xyz', ''}}}, next_msg())
     nvim('command', "call jobstop(j)")
-    eq({'notification', 'exit', {0}}, next_message())
+    eq({'notification', 'exit', {0, 0}}, next_msg())
   end)
 
   it('preserves NULs', function()
@@ -63,56 +62,64 @@ describe('jobs', function()
     file:close()
 
     -- 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', 'exit', {0}}, next_message())
+    nvim('command', "let j = jobstart(['cat', '"..filename.."'], g:job_opts)")
+    eq({'notification', 'stdout', {0, {'abc\ndef', ''}}}, next_msg())
+    eq({'notification', 'exit', {0, 0}}, next_msg())
     os.remove(filename)
 
     -- jobsend() preserves NULs.
-    nvim('command', "let j = jobstart('xxx', 'cat', ['-'])")
+    nvim('command', "let j = jobstart(['cat', '-'], g:job_opts)")
     nvim('command', [[call jobsend(j, ["123\n456",""])]])
-    eq({'notification', 'stdout', {{'123\n456', ''}}}, next_message())
+    eq({'notification', 'stdout', {0, {'123\n456', ''}}}, next_msg())
     nvim('command', "call jobstop(j)")
   end)
 
   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', "let j = jobstart(['cat', '-'], g:job_opts)")
     nvim('command', 'call jobsend(j, "abc\\nxyz")')
-    eq({'notification', 'stdout', {{'abc', 'xyz'}}}, next_message())
+    eq({'notification', 'stdout', {0, {'abc', 'xyz'}}}, next_msg())
     nvim('command', "call jobstop(j)")
-    eq({'notification', 'exit', {0}}, next_message())
+    eq({'notification', 'exit', {0, 0}}, next_msg())
   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', "let j = jobstart(['cat', '-'], g:job_opts)")
     nvim('command', 'call jobsend(j, "a\\n\\nc\\n\\n\\n\\nb\\n\\n")')
-    eq({'notification', 'stdout', {{'a', '', 'c', '', '', '', 'b', '', ''}}},
-      next_message())
+    eq({'notification', 'stdout',
+      {0, {'a', '', 'c', '', '', '', 'b', '', ''}}}, next_msg())
   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', "let j = jobstart(['cat', '-'], g:job_opts)")
     nvim('command', 'call jobsend(j, ["\n123\n", "abc\\nxyz\n", ""])')
-    eq({'notification', 'stdout', {{'\n123\n', 'abc\nxyz\n', ''}}},
-      next_message())
+    eq({'notification', 'stdout', {0, {'\n123\n', 'abc\nxyz\n', ''}}},
+      next_msg())
     nvim('command', "call jobstop(j)")
-    eq({'notification', 'exit', {0}}, next_message())
+    eq({'notification', 'exit', {0, 0}}, next_msg())
   end)
 
   it('can avoid sending final newline', function()
-    nvim('command', notify_str('v:job_data[1]', 'get(v:job_data, 2)'))
-    nvim('command', "let j = jobstart('xxx', 'cat', ['-'])")
+    nvim('command', "let j = jobstart(['cat', '-'], g:job_opts)")
     nvim('command', 'call jobsend(j, ["some data", "without\nfinal nl"])')
-    eq({'notification', 'stdout', {{'some data', 'without\nfinal nl'}}},
-      next_message())
+    eq({'notification', 'stdout', {0, {'some data', 'without\nfinal nl'}}},
+      next_msg())
     nvim('command', "call jobstop(j)")
-    eq({'notification', 'exit', {0}}, next_message())
+    eq({'notification', 'exit', {0, 0}}, next_msg())
+  end)
+
+  it('can close the job streams with jobclose', function()
+    nvim('command', "let j = jobstart(['cat', '-'], g:job_opts)")
+    nvim('command', 'call jobclose(j, "stdin")')
+    eq({'notification', 'exit', {0, 0}}, next_msg())
   end)
 
+  it('wont allow jobsend with a job that closed stdin', function()
+    nvim('command', "let j = jobstart(['cat', '-'], g:job_opts)")
+    nvim('command', 'call jobclose(j, "stdin")')
+    eq(false, pcall(function()
+      nvim('command', 'call jobsend(j, ["some data"])')
+    end))
+  end)
 
   it('will not allow jobsend/stop on a non-existent job', function()
     eq(false, pcall(eval, "jobsend(-1, 'lol')"))
@@ -120,33 +127,164 @@ describe('jobs', function()
   end)
 
   it('will not allow jobstop twice on the same job', function()
-    nvim('command', "let j = jobstart('xxx', 'cat', ['-'])")
+    nvim('command', "let j = jobstart(['cat', '-'], g:job_opts)")
     neq(0, eval('j'))
     eq(true, pcall(eval, "jobstop(j)"))
     eq(false, pcall(eval, "jobstop(j)"))
   end)
 
   it('will not cause a memory leak if we leave a job running', function()
-    nvim('command', "call jobstart('xxx', 'cat', ['-'])")
+    nvim('command', "call jobstart(['cat', '-'], g:job_opts)")
+  end)
+
+  it('can pass user data to the callback', function()
+    nvim('command', 'let g:job_opts.user = {"n": 5, "s": "str", "l": [1]}')
+    nvim('command', "call jobstart(['echo'], g:job_opts)")
+    local data = {n = 5, s = 'str', l = {1}}
+    eq({'notification', 'stdout', {data, {'', ''}}}, next_msg())
+    eq({'notification', 'exit', {data, 0}}, next_msg())
+  end)
+
+  it('can omit options', function()
+    neq(0, nvim('eval', 'delete(".Xtestjob")'))
+    nvim('command', "call jobstart(['touch', '.Xtestjob'])")
+    nvim('command', "sleep 100m")
+    eq(0, nvim('eval', 'delete(".Xtestjob")'))
+  end)
+
+  it('can omit data callbacks', function()
+    nvim('command', 'unlet g:job_opts.on_stdout')
+    nvim('command', 'unlet g:job_opts.on_stderr')
+    nvim('command', 'let g:job_opts.user = 5')
+    nvim('command', "call jobstart(['echo'], g:job_opts)")
+    eq({'notification', 'exit', {5, 0}}, next_msg())
+  end)
+
+  it('can omit exit callback', function()
+    nvim('command', 'unlet g:job_opts.on_exit')
+    nvim('command', 'let g:job_opts.user = 5')
+    nvim('command', "call jobstart(['echo'], g:job_opts)")
+    eq({'notification', 'stdout', {5, {'', ''}}}, next_msg())
+  end)
+
+  it('will pass return code with the exit event', function()
+    nvim('command', 'let g:job_opts.user = 5')
+    nvim('command', "call jobstart([&sh, '-c', 'exit 55'], g:job_opts)")
+    eq({'notification', 'exit', {5, 55}}, next_msg())
+  end)
+
+  it('can receive dictionary functions', function()
+    source([[
+    let g:dict = {'id': 10}
+    function g:dict.on_exit(id, code, event)
+      call rpcnotify(g:channel, a:event, a:code, self.id)
+    endfunction
+    call jobstart([&sh, '-c', 'exit 45'], g:dict)
+    ]])
+    eq({'notification', 'exit', {45, 10}}, next_msg())
+  end)
+
+  describe('jobwait', function()
+    it('returns a list of status codes', function()
+      source([[
+      call rpcnotify(g:channel, 'wait', jobwait([
+      \  jobstart([&sh, '-c', 'sleep 0.10; exit 4']),
+      \  jobstart([&sh, '-c', 'sleep 0.110; exit 5']),
+      \  jobstart([&sh, '-c', 'sleep 0.210; exit 6']),
+      \  jobstart([&sh, '-c', 'sleep 0.310; exit 7'])
+      \  ]))
+      ]])
+      eq({'notification', 'wait', {{4, 5, 6, 7}}}, next_msg())
+    end)
+
+    it('will run callbacks while waiting', function()
+      source([[
+      let g:dict = {'id': 10}
+      let g:l = []
+      function g:dict.on_stdout(id, data)
+        call add(g:l, a:data[0])
+      endfunction
+      call jobwait([
+      \  jobstart([&sh, '-c', 'sleep 0.010; echo 4'], g:dict),
+      \  jobstart([&sh, '-c', 'sleep 0.030; echo 5'], g:dict),
+      \  jobstart([&sh, '-c', 'sleep 0.050; echo 6'], g:dict),
+      \  jobstart([&sh, '-c', 'sleep 0.070; echo 7'], g:dict)
+      \  ])
+      call rpcnotify(g:channel, 'wait', g:l)
+      ]])
+      eq({'notification', 'wait', {{'4', '5', '6', '7'}}}, next_msg())
+    end)
+
+    it('will return status codes in the order of passed ids', function()
+      source([[
+      call rpcnotify(g:channel, 'wait', jobwait([
+      \  jobstart([&sh, '-c', 'sleep 0.070; exit 4']),
+      \  jobstart([&sh, '-c', 'sleep 0.050; exit 5']),
+      \  jobstart([&sh, '-c', 'sleep 0.030; exit 6']),
+      \  jobstart([&sh, '-c', 'sleep 0.010; exit 7'])
+      \  ]))
+      ]])
+      eq({'notification', 'wait', {{4, 5, 6, 7}}}, next_msg())
+    end)
+
+    it('will return -3 for invalid job ids', function()
+      source([[
+      call rpcnotify(g:channel, 'wait', jobwait([
+      \  -10,
+      \  jobstart([&sh, '-c', 'sleep 0.01; exit 5']),
+      \  ]))
+      ]])
+      eq({'notification', 'wait', {{-3, 5}}}, next_msg())
+    end)
+
+    it('will return -2 when interrupted', function()
+      execute('call rpcnotify(g:channel, "ready") | '..
+              'call rpcnotify(g:channel, "wait", '..
+              'jobwait([jobstart([&sh, "-c", "sleep 10; exit 55"])]))')
+      eq({'notification', 'ready', {}}, next_msg())
+      feed('<c-c>')
+      eq({'notification', 'wait', {{-2}}}, next_msg())
+    end)
+
+    describe('with timeout argument', function()
+      it('will return -1 if the wait timed out', function()
+        source([[
+        call rpcnotify(g:channel, 'wait', jobwait([
+        \  jobstart([&sh, '-c', 'sleep 0.05; exit 4']),
+        \  jobstart([&sh, '-c', 'sleep 0.3; exit 5']),
+        \  ], 100))
+        ]])
+        eq({'notification', 'wait', {{4, -1}}}, next_msg())
+      end)
+
+      it('can pass 0 to check if a job exists', function()
+        source([[
+        call rpcnotify(g:channel, 'wait', jobwait([
+        \  jobstart([&sh, '-c', 'sleep 0.05; exit 4']),
+        \  jobstart([&sh, '-c', 'sleep 0.3; exit 5']),
+        \  ], 0))
+        ]])
+        eq({'notification', 'wait', {{-1, -1}}}, next_msg())
+      end)
+    end)
   end)
 
   -- FIXME need to wait until jobsend succeeds before calling jobstop
   pending('will only emit the "exit" event after "stdout" and "stderr"', function()
-    nvim('command', notify_job())
-    nvim('command', "let j = jobstart('xxx', 'cat', ['-'])")
+    nvim('command', "let j = jobstart(['cat', '-'], g:job_opts)")
     local jobid = nvim('eval', 'j')
     nvim('eval', 'jobsend(j, "abcdef")')
     nvim('eval', 'jobstop(j)')
-    eq({'notification', 'j', {{jobid, 'stdout', {'abcdef'}}}}, next_message())
-    eq({'notification', 'j', {{jobid, 'exit'}}}, next_message())
+    eq({'notification', 'j', {0, {jobid, 'stdout', {'abcdef'}}}}, next_msg())
+    eq({'notification', 'j', {0, {jobid, 'exit'}}}, next_msg())
   end)
 
   describe('running tty-test program', function()
     local function next_chunk()
       local rv = ''
       while true do
-        local msg = next_message()
-        local data = msg[3][1]
+        local msg = next_msg()
+        local data = msg[3][2]
         for i = 1, #data do
           data[i] = data[i]:gsub('\n', '\000')
         end
@@ -166,9 +304,9 @@ describe('jobs', function()
     before_each(function() 
       -- the full path to tty-test seems to be required when running on travis.
       insert(nvim_dir .. '/tty-test')
-      nvim('command', 'let exec = expand("<cfile>:p")')
-      nvim('command', notify_str('v:job_data[1]', 'get(v:job_data, 2)'))
-      nvim('command', "let j = jobstart('xxx', exec, [], {})")
+      nvim('command', 'let g:job_opts.pty = 1')
+      nvim('command', 'let exec = [expand("<cfile>:p")]')
+      nvim('command', "let j = jobstart(exec, g:job_opts)")
       eq('tty ready', next_chunk())
     end)