jbe@486: --[[--
jbe@486: output,           -- collected data from stdout if process exited successfully
jbe@486: errmsg =          -- error message if execution failed or if process didn't exit successfully
jbe@486: execute.command{
jbe@486:   command               = { filename, arg1, arg2, ... },  -- command and arguments
jbe@486:   stdin_data            = stdin_data,                     -- optional data to be sent to process via stdin
jbe@486:   stdout_result_handler = stdout_result_handler,          -- callback receiving: stdout data, success boolean, optional error message
jbe@486:   stderr_line_handler   = stderr_line_handler,            -- callback for processing stderr line-wise
jbe@486:   exit_handler          = exit_handler,                   -- callback when process exited
jbe@486:   signal_handler        = signal_handler,                 -- callback when process terminated due to signal
jbe@486:   timeout_handler       = timeout_handler,                -- callback when process gets killed due to timeout
jbe@486:   abort_handler         = abort_handler,                  -- callback when process gets killed due to request by poll function
jbe@486:   abortable             = abortable,                      -- set to true if process shall be terminated if poll function requests termination
jbe@486:   poll                  = poll,                           -- alternative poll command with moonbridge_io.poll(...) semantics
jbe@486:   db                    = db,                             -- database handle for waiting for notifies
jbe@486:   db_notify_handler     = db_notify_handler               -- handler for database notifications which may return true to kill process
jbe@486: }
jbe@486: 
jbe@486: --]]--
jbe@486: 
jbe@486: function execute.command(args)
jbe@486: 
jbe@486:   local moonbridge_io = require("moonbridge_io")
jbe@486:   local poll = args.poll or moonbridge_io.poll
jbe@486: 
jbe@486:   local stdout_chunks, stderr_chunks = {}, {}
jbe@486: 
jbe@486:   local function return_error(errmsg)
jbe@486:     if args.stdout_result_handler then
jbe@486:       args.stdout_result_handler(table.concat(stdout_chunks), false, errmsg)
jbe@486:     end
jbe@486:     return nil, errmsg
jbe@486:   end
jbe@486: 
jbe@487:   if args.abortable then
jbe@487:     local pollready, pollmsg, pollterm = poll(nil, nil, 0, true)
jbe@487:     if pollterm then
jbe@488:       if args.abort_handler then args.abort_handler(pollmsg) end
jbe@487:       return_error(pollmsg)
jbe@487:     end
jbe@487:   end
jbe@487: 
jbe@487:   local start = moonbridge_io.timeref()
jbe@487:   local process, errmsg = moonbridge_io.exec(table.unpack(args.command))
jbe@487:   if not process then return nil, errmsg end
jbe@487: 
jbe@489:   local read_fds = {
jbe@489:     [process] = true,
jbe@489:     [process.stdout] = true,
jbe@489:     [process.stderr] = true
jbe@489:   }
jbe@489:   local write_fds = {
jbe@489:     [process.stdin] = true
jbe@489:   }
jbe@487:   if args.db then
jbe@487:     read_fds[args.db.fd] = true
jbe@487:   end
jbe@487: 
jbe@486:   local function write(...)
jbe@486:     if write_fds[process.stdin] then
jbe@486:       local buffered = process.stdin:flush_nb(...)
jbe@486:       if not buffered or buffered == 0 then
jbe@486:         process.stdin:close()
jbe@486:         write_fds[process.stdin] = nil
jbe@486:       end
jbe@486:     end
jbe@486:   end
jbe@486:   write(args.stdin_data or "")
jbe@486: 
jbe@489:   local status
jbe@489: 
jbe@486:   while
jbe@489:     not status or
jbe@486:     read_fds[process.stdout] or read_fds[process.stderr] or
jbe@486:     write_fds[process.stdin]
jbe@486:   do
jbe@486:     local timeout = args.timeout and args.timeout-moonbridge_io.timeref(start)
jbe@487:     local pollready, pollmsg, pollterm =
jbe@486:       poll(read_fds, write_fds, timeout, args.abortable)
jbe@487:     if not pollready then
jbe@489:       if not status then
jbe@489:         process:kill():wait()
jbe@489:       end
jbe@486:       if pollterm then
jbe@488:         if args.abort_handler then args.abort_handler(pollmsg) end
jbe@486:       else
jbe@486:         if args.timeout_handler then args.timeout_handler() end
jbe@486:       end
jbe@486:       return return_error(pollmsg)
jbe@486:     end
jbe@489:     if not status then
jbe@489:       status = process:wait_nb()
jbe@489:       if status then
jbe@489:         read_fds[process] = nil
jbe@489:       end
jbe@489:     end
jbe@486:     if args.db then
jbe@486:       local channel, payload, pid = db:wait(0)
jbe@486:       if channel then
jbe@486:         if args.db_notify_handler(channel, payload, pid) then
jbe@486:           process:kill():wait()
jbe@486:           return return_error("Database event received")
jbe@486:         end
jbe@486:       end
jbe@486:     end
jbe@486:     if read_fds[process.stdout] then
jbe@486:       local chunk, status = process.stdout:read_nb()
jbe@486:       if not chunk or status == "eof" then
jbe@486:         process.stdout:close()
jbe@486:         read_fds[process.stdout] = nil
jbe@486:       end
jbe@486:       if chunk and chunk ~= "" then
jbe@486:         stdout_chunks[#stdout_chunks+1] = chunk
jbe@486:       end
jbe@486:     end
jbe@486:     if read_fds[process.stderr] then
jbe@486:       local chunk, status = process.stderr:read_nb()
jbe@486:       if not chunk or status == "eof" then
jbe@486:         process.stderr:close()
jbe@486:         read_fds[process.stderr] = nil
jbe@486:       end
jbe@486:       if chunk and args.stderr_line_handler then
jbe@486:         while true do
jbe@486:           local chunk1, chunk2 = string.match(chunk, "(.-)\n(.*)")
jbe@486:           if not chunk1 then break end
jbe@486:           stderr_chunks[#stderr_chunks+1] = chunk1
jbe@486:           args.stderr_line_handler(table.concat(stderr_chunks))
jbe@486:           stderr_chunks = {}
jbe@486:           chunk = chunk2
jbe@486:         end
jbe@486:         if chunk ~= "" then
jbe@486:           stderr_chunks[#stderr_chunks+1] = chunk
jbe@486:         end
jbe@486:         if status == "eof" then
jbe@486:           local line = table.concat(stderr_chunks)
jbe@486:           if #line > 0 then args.stderr_line_handler(line) end
jbe@486:         end
jbe@486:       end
jbe@486:     end
jbe@486:     write()
jbe@486:   end
jbe@486: 
jbe@486:   if status < 0 then
jbe@486:     if args.signal_handler then
jbe@486:       args.signal_handler(-status)
jbe@486:     end
jbe@486:     return return_error("Command terminated by signal " .. -status)
jbe@486:   elseif status > 0 then
jbe@486:     if args.exit_handler then
jbe@486:       args.exit_handler(status)
jbe@486:     end
jbe@486:     return return_error("Command returned exit code " .. status)
jbe@486:   elseif args.stdout_result_handler then
jbe@486:     args.stdout_result_handler(table.concat(stdout_chunks), true)
jbe@486:     return true
jbe@486:   else
jbe@486:     return table.concat(stdout_chunks)
jbe@486:   end
jbe@486: 
jbe@486: end