merge from master

docs/api/index.rst

@@ -27,6 +27,7 @@ For those of you who want the gritty details.
     aliases
     dirstack
     jobs
+    proc
     inspectors
     completer
     shell

docs/api/proc.rst

@@ -0,0 +1,9 @@
+.. _xonsh_proc:
+
+******************************************************
+Python Procedures as Subprocess Commands (``xonsh.proc``)
+******************************************************
+
+.. automodule:: xonsh.proc
+    :members:
+    :undoc-members:

xonsh/built_ins.py

@@ -7,6 +7,7 @@ import sys
 import shlex
 import signal
 import locale
+import inspect
 import builtins
 import subprocess
 from io import TextIOWrapper, StringIO
@@ -16,13 +17,13 @@ from contextlib import contextmanager
 from collections import Sequence, MutableMapping, Iterable, namedtuple, \
     MutableSequence, MutableSet
 
-from xonsh.tools import string_types, redirect_stdout, redirect_stderr
+from xonsh.tools import string_types
 from xonsh.tools import suggest_commands, XonshError, ON_POSIX
 from xonsh.inspectors import Inspector
 from xonsh.environ import default_env
 from xonsh.aliases import DEFAULT_ALIASES, bash_aliases
 from xonsh.jobs import add_job, wait_for_active_job
-from xonsh.jobs import ProcProxy
+from xonsh.proc import ProcProxy, SimpleProcProxy
 
 ENV = None
 BUILTINS_LOADED = False
@@ -317,54 +318,6 @@ def iglobpath(s):
 WRITER_MODES = {'>': 'w', '>>': 'a'}
 
 
-def _run_callable_subproc(alias, args,
-                          captured=True,
-                          prev_proc=None,
-                          stdout=None):
-    """Helper for running callables as a subprocess."""
-    # compute stdin for callable
-    if prev_proc is None:
-        stdin = None
-    elif isinstance(prev_proc, ProcProxy):
-        stdin = prev_proc.stdout
-    else:
-        stdin = StringIO(prev_proc.communicate()[0].decode(), None)
-        stdin.seek(0)
-        stdin, _ = stdin.read(), stdin.close()
-    # Redirect the output streams temporarily. merge with possible
-    # return values from alias function.
-    if stdout is PIPE:
-        # handles captured mode
-        new_stdout, new_stderr = StringIO(), StringIO()
-        with redirect_stdout(new_stdout), redirect_stderr(new_stderr):
-            rtn = alias(args, stdin=stdin)
-        proxy_stdout = new_stdout.getvalue()
-        proxy_stderr = new_stderr.getvalue()
-        if isinstance(rtn, str):
-            proxy_stdout += rtn
-        elif isinstance(rtn, Sequence):
-            if rtn[0]:  # not None nor ''
-                proxy_stdout += rtn[0]
-            if rtn[1]:
-                proxy_stderr += rtn[1]
-        return ProcProxy(proxy_stdout, proxy_stderr)
-    else:
-        # handles uncaptured mode
-        rtn = alias(args, stdin=stdin)
-        rtnout, rtnerr = None, None
-        if isinstance(rtn, str):
-            rtnout = rtn
-            sys.stdout.write(rtn)
-        elif isinstance(rtn, Sequence):
-            if rtn[0]:
-                rtnout = rtn[0]
-                sys.stdout.write(rtn[0])
-            if rtn[1]:
-                rtnerr = rtn[1]
-                sys.stderr.write(rtn[1])
-        return ProcProxy(rtnout, rtnerr)
-
-
 RE_SHEBANG = re.compile(r'#![ \t]*(.+?)$')
 
 
@@ -474,47 +427,55 @@ def run_subproc(cmds, captured=True):
         elif alias is None:
             aliased_cmd = cmd
         elif callable(alias):
-            prev_proc = _run_callable_subproc(alias, cmd[1:],
-                                              captured=captured,
-                                              prev_proc=prev_proc,
-                                              stdout=stdout)
-            continue
+            aliased_cmd = alias
         else:
             aliased_cmd = alias + cmd[1:]
         # compute stdin for subprocess
-        prev_is_proxy = isinstance(prev_proc, ProcProxy)
         if prev_proc is None:
             stdin = None
-        elif prev_is_proxy:
-            stdin = PIPE
         else:
             stdin = prev_proc.stdout
-        subproc_kwargs = {}
-        if ON_POSIX:
-            subproc_kwargs['preexec_fn'] = _subproc_pre
-        try:
-            proc = Popen(aliased_cmd,
-                         universal_newlines=uninew,
-                         env=ENV.detype(),
-                         stdin=stdin,
-                         stdout=stdout, **subproc_kwargs)
-        except PermissionError:
-            e = 'xonsh: subprocess mode: permission denied: {0}'
-            raise XonshError(e.format(aliased_cmd[0]))
-        except FileNotFoundError:
-            cmd = aliased_cmd[0]
-            e = 'xonsh: subprocess mode: command not found: {0}'.format(cmd)
-            e += '\n' + suggest_commands(cmd, ENV, builtins.aliases)
-            raise XonshError(e)
+        if callable(aliased_cmd):
+            prev_is_proxy = True
+            numargs = len(inspect.signature(aliased_cmd).parameters)
+            if numargs == 2:
+                cls = SimpleProcProxy
+            elif numargs == 4:
+                cls = ProcProxy
+            else:
+                e = 'Expected callable with 2 or 4 arguments, not {}'
+                raise XonshError(e.format(numargs))
+            proc = cls(aliased_cmd, cmd[1:],
+                       stdin, stdout, None,
+                       universal_newlines=uninew)
+        else:
+            prev_is_proxy = False
+            subproc_kwargs = {}
+            if ON_POSIX:
+                subproc_kwargs['preexec_fn'] = _subproc_pre
+            try:
+                proc = Popen(aliased_cmd,
+                             universal_newlines=uninew,
+                             env=ENV.detype(),
+                             stdin=stdin,
+                             stdout=stdout, **subproc_kwargs)
+            except PermissionError:
+                e = 'xonsh: subprocess mode: permission denied: {0}'
+                raise XonshError(e.format(aliased_cmd[0]))
+            except FileNotFoundError:
+                cmd = aliased_cmd[0]
+                e = 'xonsh: subprocess mode: command not found: {0}'.format(cmd)
+                e += '\n' + suggest_commands(cmd, ENV, builtins.aliases)
+                raise XonshError(e)
         procs.append(proc)
         prev = None
-        if prev_is_proxy:
-            proc.stdin.write(prev_proc.stdout)
-            proc.stdin.close()
         prev_proc = proc
     for proc in procs[:-1]:
-        proc.stdout.close()
-    if not isinstance(prev_proc, ProcProxy):
+        try:
+            proc.stdout.close()
+        except OSError:
+            pass
+    if not prev_is_proxy:
         add_job({
             'cmds': cmds,
             'pids': [i.pid for i in procs],
@@ -523,16 +484,18 @@ def run_subproc(cmds, captured=True):
         })
     if background:
         return
-    wait_for_active_job()
+    if prev_is_proxy:
+        prev_proc.wait()
+    else:
+        wait_for_active_job()
     if write_target is None:
         # get output
-        if isinstance(prev_proc, ProcProxy):
-            output = prev_proc.stdout
-        elif prev_proc.stdout is not None:
+        output = ''
+        if prev_proc.stdout not in (None, sys.stdout):
             output = prev_proc.stdout.read()
         if captured:
             return output
-    elif last_stdout not in (PIPE, None):
+    elif last_stdout not in (PIPE, None, sys.stdout):
         last_stdout.close()
 
 

xonsh/jobs.py

@@ -11,10 +11,6 @@ from subprocess import TimeoutExpired
 
 from xonsh.tools import ON_WINDOWS
 
-ProcProxy = namedtuple('ProcProxy', ['stdout', 'stderr'])
-"""
-A class representing a Python function to be run as a subprocess command.
-"""
 
 
 if not ON_WINDOWS:
@@ -60,7 +56,7 @@ def _clear_dead_jobs():
     to_remove = set()
     for num, job in builtins.__xonsh_all_jobs__.items():
         obj = job['obj']
-        if isinstance(obj, ProcProxy) or obj.poll() is not None:
+        if obj.poll() is not None:
             to_remove.add(num)
     for i in to_remove:
         del builtins.__xonsh_all_jobs__[i]
@@ -136,8 +132,6 @@ def wait_for_active_job():
         return
     job = builtins.__xonsh_all_jobs__[act]
     obj = job['obj']
-    if isinstance(obj, ProcProxy):
-        return
     if job['bg']:
         return
     if ON_WINDOWS:

xonsh/proc.py

@@ -0,0 +1,306 @@
+"""Interface for running Python functions as subprocess-mode commands.
+
+Code for several helper methods in the `ProcProxy` class have been reproduced
+without modification from `subprocess.py` in the Python 3.4.2 standard library.
+The contents of `subprocess.py` (and, thus, the reproduced methods) are
+Copyright (c) 2003-2005 by Peter Astrand <astrand@lysator.liu.se> and were
+licensed to the Python Software foundation under a Contributor Agreement.
+"""
+import io
+import os
+import sys
+import platform
+
+from threading import Thread
+from subprocess import Popen, PIPE, DEVNULL, STDOUT
+from collections import Sequence
+
+from xonsh.tools import redirect_stdout, redirect_stderr, ON_WINDOWS
+
+if ON_WINDOWS:
+    import _winapi
+    import msvcrt
+
+
+class ProcProxy(Thread):
+    """
+    Class representing a function to be run as a subprocess-mode command.
+    """
+    def __init__(self, f, args,
+                 stdin=None,
+                 stdout=None,
+                 stderr=None,
+                 universal_newlines=False):
+        """Parameters
+        ----------
+        f : function
+            The function to be executed.
+        args : list
+            A (possibly empty) list containing the arguments that were given on
+            the command line
+        stdin : file-like, optional
+            A file-like object representing stdin (input can be read from
+            here).  If `stdin` is not provided or if it is explicitly set to
+            `None`, then an instance of `io.StringIO` representing an empty
+            file is used.
+        stdout : file-like, optional
+            A file-like object representing stdout (normal output can be
+            written here).  If `stdout` is not provided or if it is explicitly
+            set to `None`, then `sys.stdout` is used.
+        stderr : file-like, optional
+            A file-like object representing stderr (error output can be
+            written here).  If `stderr` is not provided or if it is explicitly
+            set to `None`, then `sys.stderr` is used.
+        """
+        self.f = f
+        """
+        The function to be executed.  It should be a function of four
+        arguments, described below.
+
+        Parameters
+        ----------
+        args : list
+            A (possibly empty) list containing the arguments that were given on
+            the command line
+        stdin : file-like
+            A file-like object representing stdin (input can be read from
+            here).
+        stdout : file-like
+            A file-like object representing stdout (normal output can be
+            written here).
+        stderr : file-like
+            A file-like object representing stderr (error output can be
+            written here).
+        """
+        self.args = args
+        self.pid = None
+        self.returncode = None
+        self.wait = self.join
+
+        handles = self._get_handles(stdin, stdout, stderr)
+        (self.p2cread, self.p2cwrite,
+         self.c2pread, self.c2pwrite,
+         self.errread, self.errwrite) = handles
+
+        # default values
+        self.stdin = stdin
+        self.stdout = None
+        self.stderr = None
+
+        if self.p2cwrite != -1:
+            self.stdin = io.open(self.p2cwrite, 'wb', -1)
+            if universal_newlines:
+                self.stdin = io.TextIOWrapper(self.stdin, write_through=True,
+                                              line_buffering=False)
+        if self.c2pread != -1:
+            self.stdout = io.open(self.c2pread, 'rb', -1)
+            if universal_newlines:
+                self.stdout = io.TextIOWrapper(self.stdout)
+        if self.errread != -1:
+            self.stderr = io.open(self.errread, 'rb', -1)
+            if universal_newlines:
+                self.stderr = io.TextIOWrapper(self.stderr)
+
+        Thread.__init__(self)
+        self.start()
+
+    def run(self):
+        """Set up input/output streams and execute the child function in a new
+        thread.  This is part of the `threading.Thread` interface and should
+        not be called directly."""
+        if self.f is None:
+            return
+        if self.stdin is not None:
+            sp_stdin = io.TextIOWrapper(self.stdin)
+        else:
+            sp_stdin = io.StringIO("")
+        if self.c2pwrite != -1:
+            sp_stdout = io.TextIOWrapper(io.open(self.c2pwrite, 'wb', -1))
+        else:
+            sp_stdout = sys.stdout
+        if self.errwrite != -1:
+            sp_stderr = io.TextIOWrapper(io.open(self.errwrite, 'wb', -1))
+        else:
+            sp_stderr = sys.stderr
+        r = self.f(self.args, sp_stdin, sp_stdout, sp_stderr)
+        self.returncode = r if r is not None else True
+
+    def poll(self):
+        """Check if the function has completed.
+
+        :return: `None` if the function is still executing, `True` if the
+                 function finished successfully, and `False` if there was an
+                 error
+        """
+        return self.returncode
+
+    # The code below (_get_devnull, _get_handles, and _make_inheritable) comes
+    # from subprocess.py in the Python 3.4.2 Standard Library
+    def _get_devnull(self):
+        if not hasattr(self, '_devnull'):
+            self._devnull = os.open(os.devnull, os.O_RDWR)
+        return self._devnull
+
+    if ON_WINDOWS:
+        def _make_inheritable(self, handle):
+            """Return a duplicate of handle, which is inheritable"""
+            h = _winapi.DuplicateHandle(
+                _winapi.GetCurrentProcess(), handle,
+                _winapi.GetCurrentProcess(), 0, 1,
+                _winapi.DUPLICATE_SAME_ACCESS)
+            return Handle(h)
+
+        def _get_handles(self, stdin, stdout, stderr):
+            """Construct and return tuple with IO objects:
+            p2cread, p2cwrite, c2pread, c2pwrite, errread, errwrite
+            """
+            if stdin is None and stdout is None and stderr is None:
+                return (-1, -1, -1, -1, -1, -1)
+
+            p2cread, p2cwrite = -1, -1
+            c2pread, c2pwrite = -1, -1
+            errread, errwrite = -1, -1
+
+            if stdin is None:
+                p2cread = _winapi.GetStdHandle(_winapi.STD_INPUT_HANDLE)
+                if p2cread is None:
+                    p2cread, _ = _winapi.CreatePipe(None, 0)
+                    p2cread = Handle(p2cread)
+                    _winapi.CloseHandle(_)
+            elif stdin == PIPE:
+                p2cread, p2cwrite = _winapi.CreatePipe(None, 0)
+                p2cread, p2cwrite = Handle(p2cread), Handle(p2cwrite)
+            elif stdin == DEVNULL:
+                p2cread = msvcrt.get_osfhandle(self._get_devnull())
+            elif isinstance(stdin, int):
+                p2cread = msvcrt.get_osfhandle(stdin)
+            else:
+                # Assuming file-like object
+                p2cread = msvcrt.get_osfhandle(stdin.fileno())
+            p2cread = self._make_inheritable(p2cread)
+
+            if stdout is None:
+                c2pwrite = _winapi.GetStdHandle(_winapi.STD_OUTPUT_HANDLE)
+                if c2pwrite is None:
+                    _, c2pwrite = _winapi.CreatePipe(None, 0)
+                    c2pwrite = Handle(c2pwrite)
+                    _winapi.CloseHandle(_)
+            elif stdout == PIPE:
+                c2pread, c2pwrite = _winapi.CreatePipe(None, 0)
+                c2pread, c2pwrite = Handle(c2pread), Handle(c2pwrite)
+            elif stdout == DEVNULL:
+                c2pwrite = msvcrt.get_osfhandle(self._get_devnull())
+            elif isinstance(stdout, int):
+                c2pwrite = msvcrt.get_osfhandle(stdout)
+            else:
+                # Assuming file-like object
+                c2pwrite = msvcrt.get_osfhandle(stdout.fileno())
+            c2pwrite = self._make_inheritable(c2pwrite)
+
+            if stderr is None:
+                errwrite = _winapi.GetStdHandle(_winapi.STD_ERROR_HANDLE)
+                if errwrite is None:
+                    _, errwrite = _winapi.CreatePipe(None, 0)
+                    errwrite = Handle(errwrite)
+                    _winapi.CloseHandle(_)
+            elif stderr == PIPE:
+                errread, errwrite = _winapi.CreatePipe(None, 0)
+                errread, errwrite = Handle(errread), Handle(errwrite)
+            elif stderr == STDOUT:
+                errwrite = c2pwrite
+            elif stderr == DEVNULL:
+                errwrite = msvcrt.get_osfhandle(self._get_devnull())
+            elif isinstance(stderr, int):
+                errwrite = msvcrt.get_osfhandle(stderr)
+            else:
+                # Assuming file-like object
+                errwrite = msvcrt.get_osfhandle(stderr.fileno())
+            errwrite = self._make_inheritable(errwrite)
+
+            return (p2cread, p2cwrite,
+                    c2pread, c2pwrite,
+                    errread, errwrite)
+
+    else:
+        # POSIX versions
+        def _get_handles(self, stdin, stdout, stderr):
+            """Construct and return tuple with IO objects:
+            p2cread, p2cwrite, c2pread, c2pwrite, errread, errwrite
+            """
+            p2cread, p2cwrite = -1, -1
+            c2pread, c2pwrite = -1, -1
+            errread, errwrite = -1, -1
+
+            if stdin is None:
+                pass
+            elif stdin == PIPE:
+                p2cread, p2cwrite = os.pipe()
+            elif stdin == DEVNULL:
+                p2cread = self._get_devnull()
+            elif isinstance(stdin, int):
+                p2cread = stdin
+            else:
+                # Assuming file-like object
+                p2cread = stdin.fileno()
+
+            if stdout is None:
+                pass
+            elif stdout == PIPE:
+                c2pread, c2pwrite = os.pipe()
+            elif stdout == DEVNULL:
+                c2pwrite = self._get_devnull()
+            elif isinstance(stdout, int):
+                c2pwrite = stdout
+            else:
+                # Assuming file-like object
+                c2pwrite = stdout.fileno()
+
+            if stderr is None:
+                pass
+            elif stderr == PIPE:
+                errread, errwrite = os.pipe()
+            elif stderr == STDOUT:
+                errwrite = c2pwrite
+            elif stderr == DEVNULL:
+                errwrite = self._get_devnull()
+            elif isinstance(stderr, int):
+                errwrite = stderr
+            else:
+                # Assuming file-like object
+                errwrite = stderr.fileno()
+
+            return (p2cread, p2cwrite,
+                    c2pread, c2pwrite,
+                    errread, errwrite)
+
+
+class SimpleProcProxy(ProcProxy):
+    """
+    Variant of `ProcProxy` for simpler functions.
+
+    The function passed into the initializer for `SimpleProcProxy` should have
+    the form described in the xonsh tutorial.  This function is then wrapped to
+    make a new function of the form expected by `ProcProxy`.
+    """
+    def __init__(self, f, args, stdin=None, stdout=None, stderr=None,
+                 universal_newlines=False):
+        def wrapped_simple_command(args, stdin, stdout, stderr):
+            try:
+                i = stdin.read()
+                with redirect_stdout(stdout), redirect_stderr(stderr):
+                    r = f(args, i)
+                if isinstance(r, str):
+                    stdout.write(r)
+                elif isinstance(r, Sequence):
+                    if r[0] is not None:
+                        stdout.write(r[0])
+                    if r[1] is not None:
+                        stderr.write(r[1])
+                elif r is not None:
+                    stdout.write(str(r))
+                return True
+            except:
+                return False
+        super().__init__(wrapped_simple_command,
+                         args, stdin, stdout, stderr,
+                         universal_newlines)

xonsh/tools.py

@@ -33,8 +33,8 @@ else:
 DEFAULT_ENCODING = sys.getdefaultencoding()
 
 ON_WINDOWS = (platform.system() == 'Windows')
-ON_MAC     = (platform.system() == 'Darwin')
-ON_POSIX   = (os.name == 'posix')
+ON_MAC = (platform.system() == 'Darwin')
+ON_POSIX = (os.name == 'posix')
 
 
 class XonshError(Exception):