Refactor jobs into a Vim version agnostic API which can be used for other purposes

2 files changed, 263 insertions, 234 deletions

diff --git a/autoload/ale/engine.vim b/autoload/ale/engine.vim
index d431088a..299d37df 100644
--- a/autoload/ale/engine.vim
+++ b/autoload/ale/engine.vim
@@ -26,21 +26,6 @@ function! s:IsExecutable(executable) abort
     return  0
 endfunction
 
-function! ale#engine#ParseVim8ProcessID(job_string) abort
-    return matchstr(a:job_string, '\d\+') + 0
-endfunction
-
-function! s:GetJobID(job) abort
-    if has('nvim')
-        "In NeoVim, job values are just IDs.
-        return a:job
-    endif
-
-    " For Vim 8, the job is a different variable type, and we can parse the
-    " process ID from the string.
-    return ale#engine#ParseVim8ProcessID(string(a:job))
-endfunction
-
 function! ale#engine#InitBufferInfo(buffer) abort
     if !has_key(g:ale_buffer_info, a:buffer)
         " job_list will hold the list of jobs
@@ -63,84 +48,17 @@ function! ale#engine#InitBufferInfo(buffer) abort
     endif
 endfunction
 
-" A map from timer IDs to Vim 8 jobs, for tracking jobs that need to be killed
-" with SIGKILL if they don't terminate right away.
-let s:job_kill_timers = {}
-
-" Check if a job is still running, in either Vim version.
-function! s:IsJobRunning(job) abort
-    if has('nvim')
-        try
-            " In NeoVim, if the job isn't running, jobpid() will throw.
-            call jobpid(a:job)
-            return 1
-        catch
-        endtry
-
-        return 0
-    endif
-
-    return job_status(a:job) ==# 'run'
-endfunction
-
-function! s:KillHandler(timer) abort
-    let l:job = remove(s:job_kill_timers, a:timer)
-
-    " For NeoVim, we have to send SIGKILL ourselves manually, as NeoVim
-    " doesn't do it properly.
-    if has('nvim')
-        let l:pid = 0
-
-        " We can fail to get the PID here if the job manages to stop already.
-        try
-            let l:pid = jobpid(l:job)
-        catch
-        endtry
-
-        if l:pid > 0
-            if has('win32')
-                " Windows
-                call system('taskkill /pid ' . l:pid . ' /f')
-            else
-                " Linux, Mac OSX, etc.
-                call system('kill -9 ' . l:pid)
-            endif
-        endif
-    else
-        call job_stop(l:job, 'kill')
-    endif
-endfunction
-
-function! ale#engine#ClearJob(job) abort
+function! ale#engine#ClearJob(job_id) abort
     if get(g:, 'ale_run_synchronously') == 1
-        call remove(s:job_info_map, a:job)
+        call remove(s:job_info_map, a:job_id)
 
         return
     endif
 
-    let l:job_id = s:GetJobID(a:job)
-
-    if has('nvim')
-        call jobstop(a:job)
-    else
-        " We must close the channel for reading the buffer if it is open
-        " when stopping a job. Otherwise, we will get errors in the status line.
-        if ch_status(job_getchannel(a:job)) ==# 'open'
-            call ch_close_in(job_getchannel(a:job))
-        endif
+    call ale#job#Stop(a:job_id)
 
-        " Ask nicely for the job to stop.
-        call job_stop(a:job)
-    endif
-
-    " If a job doesn't stop immediately, queue a timer which will
-    " send SIGKILL to the job, if it's alive by the time the timer ticks.
-    if s:IsJobRunning(a:job)
-        let s:job_kill_timers[timer_start(100, function('s:KillHandler'))] = a:job
-    endif
-
-    if has_key(s:job_info_map, l:job_id)
-        call remove(s:job_info_map, l:job_id)
+    if has_key(s:job_info_map, a:job_id)
+        call remove(s:job_info_map, a:job_id)
     endif
 endfunction
 
@@ -152,16 +70,14 @@ function! s:StopPreviousJobs(buffer, linter) abort
 
     let l:new_job_list = []
 
-    for l:job in g:ale_buffer_info[a:buffer].job_list
-        let l:job_id = s:GetJobID(l:job)
-
+    for l:job_id in g:ale_buffer_info[a:buffer].job_list
         if has_key(s:job_info_map, l:job_id)
         \&& s:job_info_map[l:job_id].linter.name ==# a:linter.name
             " Stop jobs which match the buffer and linter.
-            call ale#engine#ClearJob(l:job)
+            call ale#engine#ClearJob(l:job_id)
         else
             " Keep other jobs in the list.
-            call add(l:new_job_list, l:job)
+            call add(l:new_job_list, l:job_id)
         endif
     endfor
 
@@ -169,41 +85,6 @@ function! s:StopPreviousJobs(buffer, linter) abort
     let g:ale_buffer_info[a:buffer].job_list = l:new_job_list
 endfunction
 
-function! s:GatherOutputVim(channel, data) abort
-    let l:job_id = s:GetJobID(ch_getjob(a:channel))
-
-    if !has_key(s:job_info_map, l:job_id)
-        return
-    endif
-
-    call add(s:job_info_map[l:job_id].output, a:data)
-endfunction
-
-function! s:GatherOutputNeoVim(job, data, event) abort
-    let l:job_id = s:GetJobID(a:job)
-
-    if !has_key(s:job_info_map, l:job_id)
-        return
-    endif
-
-    " Join the lines passed to ale, because Neovim splits them up.
-    " a:data is a list of strings, where every item is a new line, except the
-    " first one, which is the continuation of the last item passed last time.
-    call ale#engine#JoinNeovimOutput(s:job_info_map[l:job_id].output, a:data)
-endfunction
-
-function! ale#engine#JoinNeovimOutput(output, data) abort
-    if empty(a:output)
-        call extend(a:output, a:data)
-    else
-        " Extend the previous line, which can be continued.
-        let a:output[-1] .= get(a:data, 0, '')
-
-        " Add the new lines.
-        call extend(a:output, a:data[1:])
-    endif
-endfunction
-
 " Register a temporary file to be managed with the ALE engine for
 " a current job run.
 function! ale#engine#ManageFile(buffer, filename) abort
@@ -255,24 +136,27 @@ function! ale#engine#RemoveManagedFiles(buffer) abort
     let g:ale_buffer_info[a:buffer].temporary_directory_list = []
 endfunction
 
-function! s:HandleExit(job) abort
-    if a:job ==# 'no process'
-        " Stop right away when the job is not valid in Vim 8.
-        return
+function! s:GatherOutput(job_id, line) abort
+    if has_key(s:job_info_map, a:job_id)
+        call add(s:job_info_map[a:job_id].output, a:line)
     endif
+endfunction
 
-    let l:job_id = s:GetJobID(a:job)
-
-    if !has_key(s:job_info_map, l:job_id)
+function! s:HandleExit(job_id, exit_code) abort
+    if !has_key(s:job_info_map, a:job_id)
         return
     endif
 
-    let l:job_info = s:job_info_map[l:job_id]
+    let l:job_info = s:job_info_map[a:job_id]
     let l:linter = l:job_info.linter
     let l:output = l:job_info.output
     let l:buffer = l:job_info.buffer
     let l:next_chain_index = l:job_info.next_chain_index
 
+    if g:ale_history_enabled
+        call ale#history#SetExitCode(l:buffer, a:job_id, a:exit_code)
+    endif
+
     " Call the same function for stopping jobs again to clean up the job
     " which just closed.
     call s:StopPreviousJobs(l:buffer, l:linter)
@@ -294,7 +178,7 @@ function! s:HandleExit(job) abort
 
     " Log the output of the command for ALEInfo if we should.
     if g:ale_history_enabled && g:ale_history_log_output
-        call ale#history#RememberOutput(l:buffer, l:job_id, l:output[:])
+        call ale#history#RememberOutput(l:buffer, a:job_id, l:output[:])
     endif
 
     let l:linter_loclist = ale#util#GetFunction(l:linter.callback)(l:buffer, l:output)
@@ -368,36 +252,6 @@ function! ale#engine#SetResults(buffer, loclist) abort
     endif
 endfunction
 
-function! s:SetExitCode(job, exit_code) abort
-    let l:job_id = s:GetJobID(a:job)
-
-    if !has_key(s:job_info_map, l:job_id)
-        return
-    endif
-
-    let l:buffer = s:job_info_map[l:job_id].buffer
-
-    call ale#history#SetExitCode(l:buffer, l:job_id, a:exit_code)
-endfunction
-
-function! s:HandleExitNeoVim(job, exit_code, event) abort
-    if g:ale_history_enabled
-        call s:SetExitCode(a:job, a:exit_code)
-    endif
-
-    call s:HandleExit(a:job)
-endfunction
-
-function! s:HandleExitVim(channel) abort
-    call s:HandleExit(ch_getjob(a:channel))
-endfunction
-
-" Vim returns the exit status with one callback,
-" and the channel will close later in another callback.
-function! s:HandleExitStatusVim(job, exit_code) abort
-    call s:SetExitCode(a:job, a:exit_code)
-endfunction
-
 function! ale#engine#FixLocList(buffer, linter, loclist) abort
     let l:new_loclist = []
 
@@ -542,85 +396,51 @@ function! s:RunJob(options) abort
         let l:read_buffer = 0
     endif
 
-    if !has('nvim')
-        " The command will be executed in a subshell. This fixes a number of
-        " issues, including reading the PATH variables correctly, %PATHEXT%
-        " expansion on Windows, etc.
-        "
-        " NeoVim handles this issue automatically if the command is a String.
-        let l:command = has('win32')
-        \   ?  'cmd /c ' . l:command
-        \   : split(&shell) + split(&shellcmdflag) + [l:command]
+    " The command will be executed in a subshell. This fixes a number of
+    " issues, including reading the PATH variables correctly, %PATHEXT%
+    " expansion on Windows, etc.
+    "
+    " NeoVim handles this issue automatically if the command is a String,
+    " but we'll do this explicitly, so we use thes same exact command for both
+    " versions.
+    let l:command = has('win32')
+    \   ?  'cmd /c ' . l:command
+    \   : split(&shell) + split(&shellcmdflag) + [l:command]
+
+    let l:job_options = {
+    \   'mode': 'nl',
+    \   'exit_cb': function('s:HandleExit'),
+    \}
+
+    if l:output_stream ==# 'stderr'
+        let l:job_options.err_cb = function('s:GatherOutput')
+    elseif l:output_stream ==# 'both'
+        let l:job_options.out_cb = function('s:GatherOutput')
+        let l:job_options.err_cb = function('s:GatherOutput')
+    else
+        let l:job_options.out_cb = function('s:GatherOutput')
     endif
 
     if get(g:, 'ale_run_synchronously') == 1
         " Find a unique Job value to use, which will be the same as the ID for
         " running commands synchronously. This is only for test code.
-        let l:job = len(s:job_info_map) + 1
+        let l:job_id = len(s:job_info_map) + 1
 
-        while has_key(s:job_info_map, l:job)
-            let l:job += 1
+        while has_key(s:job_info_map, l:job_id)
+            let l:job_id += 1
         endwhile
-    elseif has('nvim')
-        if l:output_stream ==# 'stderr'
-            " Read from stderr instead of stdout.
-            let l:job = jobstart(l:command, {
-            \   'on_stderr': function('s:GatherOutputNeoVim'),
-            \   'on_exit': function('s:HandleExitNeoVim'),
-            \})
-        elseif l:output_stream ==# 'both'
-            let l:job = jobstart(l:command, {
-            \   'on_stdout': function('s:GatherOutputNeoVim'),
-            \   'on_stderr': function('s:GatherOutputNeoVim'),
-            \   'on_exit': function('s:HandleExitNeoVim'),
-            \})
-        else
-            let l:job = jobstart(l:command, {
-            \   'on_stdout': function('s:GatherOutputNeoVim'),
-            \   'on_exit': function('s:HandleExitNeoVim'),
-            \})
-        endif
     else
-        let l:job_options = {
-        \   'in_mode': 'nl',
-        \   'out_mode': 'nl',
-        \   'err_mode': 'nl',
-        \   'close_cb': function('s:HandleExitVim'),
-        \}
-
-        if g:ale_history_enabled
-            " We only need to capture the exit status if we are going to
-            " save it in the history. Otherwise, we don't care.
-            let l:job_options.exit_cb = function('s:HandleExitStatusVim')
-        endif
-
-        if l:output_stream ==# 'stderr'
-            " Read from stderr instead of stdout.
-            let l:job_options.err_cb = function('s:GatherOutputVim')
-        elseif l:output_stream ==# 'both'
-            " Read from both streams.
-            let l:job_options.out_cb = function('s:GatherOutputVim')
-            let l:job_options.err_cb = function('s:GatherOutputVim')
-        else
-            let l:job_options.out_cb = function('s:GatherOutputVim')
-        endif
-
-        " Vim 8 will read the stdin from the file's buffer.
-        let l:job = job_start(l:command, l:job_options)
+        let l:job_id = ale#job#Start(l:command, l:job_options)
     endif
 
     let l:status = 'failed'
-    let l:job_id = 0
 
     " Only proceed if the job is being run.
-    if has('nvim')
-    \ || get(g:, 'ale_run_synchronously') == 1
-    \ || (l:job !=# 'no process' && job_status(l:job) ==# 'run')
+    if l:job_id
         " Add the job to the list of jobs, so we can track them.
-        call add(g:ale_buffer_info[l:buffer].job_list, l:job)
+        call add(g:ale_buffer_info[l:buffer].job_list, l:job_id)
 
         let l:status = 'started'
-        let l:job_id = s:GetJobID(l:job)
         " Store the ID for the job in the map to read back again.
         let s:job_info_map[l:job_id] = {
         \   'linter': l:linter,
@@ -643,7 +463,9 @@ function! s:RunJob(options) abort
         \   ?  join(l:command[0:1]) . ' ' . ale#Escape(l:command[2])
         \   : l:command
         \)
-        call s:HandleExit(l:job)
+
+        " TODO, get the exit system of the shell call and pass it on here.
+        call l:job_options.exit_cb(l:job_id, 0)
     endif
 endfunction
 
@@ -793,8 +615,8 @@ function! ale#engine#WaitForJobs(deadline) abort
     while l:should_wait_more
         let l:should_wait_more = 0
 
-        for l:job in l:job_list
-            if job_status(l:job) ==# 'run'
+        for l:job_id in l:job_list
+            if ale#job#IsRunning(l:job_id)
                 let l:now = ale#util#ClockMilliseconds()
 
                 if l:now - l:start_time > a:deadline
@@ -822,8 +644,8 @@ function! ale#engine#WaitForJobs(deadline) abort
 
     " Check again to see if any jobs are running.
     for l:info in values(g:ale_buffer_info)
-        for l:job in l:info.job_list
-            if job_status(l:job) ==# 'run'
+        for l:job_id in l:info.job_list
+            if ale#job#IsRunning(l:job_id)
                 let l:has_new_jobs = 1
                 break
             endif
diff --git a/autoload/ale/job.vim b/autoload/ale/job.vim
new file mode 100644
index 00000000..a9965444
--- /dev/null
+++ b/autoload/ale/job.vim
@@ -0,0 +1,207 @@
+" Author: w0rp <devw0rp@gmail.com>
+" Deciption: APIs for working with Asynchronous jobs, with an API normalised
+" between Vim 8 and NeoVim.
+"
+" Important functions are described below. They are:
+"
+"   ale#job#Start(command, options) -> job_id
+"   ale#job#IsRunning(job_id) -> 1 if running, 0 otherwise.
+"   ale#job#Stop(job_id)
+
+let s:job_map = {}
+" A map from timer IDs to jobs, for tracking jobs that need to be killed
+" with SIGKILL if they don't terminate right away.
+let s:job_kill_timers = {}
+
+function! s:KillHandler(timer) abort
+    let l:job = remove(s:job_kill_timers, a:timer)
+    call job_stop(l:job, 'kill')
+endfunction
+
+function! ale#job#JoinNeovimOutput(output, data) abort
+    if empty(a:output)
+        call extend(a:output, a:data)
+    else
+        " Extend the previous line, which can be continued.
+        let a:output[-1] .= get(a:data, 0, '')
+
+        " Add the new lines.
+        call extend(a:output, a:data[1:])
+    endif
+endfunction
+
+" Note that jobs and IDs are the same thing on NeoVim.
+function! s:HandleNeoVimLines(job, callback, output, data) abort
+    call ale#job#JoinNeovimOutput(a:output, a:data)
+
+    for l:line in a:output
+        call a:callback(a:job, l:line)
+    endfor
+endfunction
+
+function! s:NeoVimCallback(job, data, event) abort
+    let l:job_info = s:job_map[a:job]
+
+    if a:event ==# 'stdout'
+        call s:HandleNeoVimLines(
+        \   a:job,
+        \   ale#util#GetFunction(l:job_info.out_cb),
+        \   l:job_info.out_cb_output,
+        \   a:data,
+        \)
+    elseif a:event ==# 'stderr'
+        call s:HandleNeoVimLines(
+        \   a:job,
+        \   ale#util#GetFunction(l:job_info.err_cb),
+        \   l:job_info.err_cb_output,
+        \   a:data,
+        \)
+    else
+        call ale#util#GetFunction(l:job_info.exit_cb)(a:job, a:data)
+    endif
+endfunction
+
+function! s:VimOutputCallback(channel, data) abort
+    let l:job = ch_getjob(a:channel)
+    let l:job_id = ale#job#ParseVim8ProcessID(string(l:job))
+    call ale#util#GetFunction(s:job_map[l:job_id].out_cb)(l:job_id, a:data)
+endfunction
+
+function! s:VimErrorCallback(channel, data) abort
+    let l:job = ch_getjob(a:channel)
+    let l:job_id = ale#job#ParseVim8ProcessID(string(l:job))
+    call ale#util#GetFunction(s:job_map[l:job_id].err_cb)(l:job_id, a:data)
+endfunction
+
+function! s:VimCloseCallback(channel) abort
+    " Call job_status, which will trigger the exit callback below.
+    " This behaviour is described in :help job-status
+    call job_status(ch_getjob(a:channel))
+endfunction
+
+function! s:VimExitCallback(job, exit_code) abort
+    let l:job_id = ale#job#ParseVim8ProcessID(string(a:job))
+    call ale#util#GetFunction(s:job_map[l:job_id].exit_cb)(l:job_id, a:exit_code)
+endfunction
+
+function! ale#job#ParseVim8ProcessID(job_string) abort
+    return matchstr(a:job_string, '\d\+') + 0
+endfunction
+
+function! ale#job#ValidateArguments(command, options) abort
+    if a:options.mode !=# 'nl' && a:options.mode !=# 'raw'
+        throw 'Invalid mode: ' . a:options.mode
+    endif
+endfunction
+
+" Start a job with options which are agnostic to Vim and NeoVim.
+"
+" The following options are accepted:
+"
+" out_cb  - A callback for receiving stdin.  Arguments: (job_id, data)
+" err_cb  - A callback for receiving stderr. Arguments: (job_id, data)
+" exit_cb - A callback for program exit.     Arguments: (job_id, status_code)
+" mode    - A mode for I/O. Can be 'nl' for split lines or 'raw'.
+function! ale#job#Start(command, options) abort
+    call ale#job#ValidateArguments(a:command, a:options)
+
+    let l:job_info = copy(a:options)
+    let l:job_options = {}
+
+    if has('nvim')
+        if has_key(a:options, 'out_cb')
+            let l:job_options.on_stdout = function('s:NeoVimCallback')
+            let l:job_info.out_cb_output = []
+        endif
+
+        if has_key(a:options, 'err_cb')
+            let l:job_options.on_stderr = function('s:NeoVimCallback')
+            let l:job_info.err_cb_output = []
+        endif
+
+        if has_key(a:options, 'exit_cb')
+            let l:job_options.on_exit = function('s:NeoVimCallback')
+        endif
+
+        let l:job_info.job = jobstart(a:command, l:job_options)
+        let l:job_id = l:job_info.job
+    else
+        let l:job_options = {
+        \   'in_mode': l:job_info.mode,
+        \   'out_mode': l:job_info.mode,
+        \   'err_mode': l:job_info.mode,
+        \}
+
+        if has_key(a:options, 'out_cb')
+            let l:job_options.out_cb = function('s:VimOutputCallback')
+        endif
+
+        if has_key(a:options, 'err_cb')
+            let l:job_options.err_cb = function('s:VimErrorCallback')
+        endif
+
+        if has_key(a:options, 'exit_cb')
+            " Set a close callback to which simply calls job_status()
+            " when the channel is closed, which can trigger the exit callback
+            " earlier on.
+            let l:job_options.close_cb = function('s:VimCloseCallback')
+            let l:job_options.exit_cb = function('s:VimExitCallback')
+        endif
+
+        " Vim 8 will read the stdin from the file's buffer.
+        let l:job_info.job = job_start(a:command, l:job_options)
+        let l:job_id = ale#job#ParseVim8ProcessID(string(l:job_info.job))
+    endif
+
+    if l:job_id
+        " Store the job in the map for later only if we can get the ID.
+        let s:job_map[l:job_id] = l:job_info
+    endif
+
+    return l:job_id
+endfunction
+
+" Given a job ID, return 1 if the job is currently running.
+" Invalid job IDs will be ignored.
+function! ale#job#IsRunning(job_id) abort
+    if has('nvim')
+        try
+            " In NeoVim, if the job isn't running, jobpid() will throw.
+            call jobpid(a:job_id)
+            return 1
+        catch
+        endtry
+    elseif has_key(s:job_map, a:job_id)
+        let l:job = s:job_map[a:job_id].job
+        return job_status(l:job) ==# 'run'
+    endif
+
+    return 0
+endfunction
+
+" Given a Job ID, stop that job.
+" Invalid job IDs will be ignored.
+function! ale#job#Stop(job_id) abort
+    if has('nvim')
+        " FIXME: NeoVim kills jobs on a timer, but will not kill any processes
+        " which are child processes on Unix. Some work needs to be done to
+        " kill child processes to stop long-running processes like pylint.
+        call jobstop(a:job_id)
+    elseif has_key(s:job_map, a:job_id)
+        let l:job = s:job_map[a:job_id].job
+
+        " We must close the channel for reading the buffer if it is open
+        " when stopping a job. Otherwise, we will get errors in the status line.
+        if ch_status(job_getchannel(l:job)) ==# 'open'
+            call ch_close_in(job_getchannel(l:job))
+        endif
+
+        " Ask nicely for the job to stop.
+        call job_stop(l:job)
+
+        if ale#job#IsRunning(l:job)
+            " Set a 100ms delay for killing the job with SIGKILL.
+            let s:job_kill_timers[timer_start(100, function('s:KillHandler'))] = l:job
+        endif
+    endif
+endfunction