Merge pull request #1893 from xonsh/pipe

Pipe fixes(?)

news/rs.rst

@@ -51,6 +51,6 @@
 * Added a minimum time buffer time for command pipelines to check for
   if previous commands have executed successfully.  This is helpful
   for pipelines where the last command takes a long time to start up,
-  such as GNU Parallel.
+  such as GNU Parallel. This also checks to make sure that output has occured.
 
 **Security:** None

tests/bin/cat

@@ -0,0 +1,5 @@
+#!/usr/bin/env python
+import sys
+with open(sys.argv[-1]) as f:
+    for line in f:
+        print(line, end='')

tests/bin/cat.bat

@@ -0,0 +1,15 @@
+@echo on
+call :s_which py.exe
+rem note that %~dp0 is dir of this batch script
+if not "%_path%" == "" (
+  py -3 %~dp0cat %*
+) else (
+  python %~dp0cat %*
+)
+
+goto :eof
+
+:s_which
+  setlocal
+  endlocal & set _path=%~$PATH:1
+  goto :eof

tests/bin/wc

@@ -0,0 +1,14 @@
+#!/usr/bin/env python
+import sys
+
+if len(sys.argv) == 1:
+    f = sys.stdin.buffer
+else:
+    f = open(sys.argv[1], 'rb')
+
+doc = f.read()
+lines = len(doc.splitlines())
+words = len(doc.split())
+bytes = len(doc)
+
+print(' {0} {1:>2} {2} {3}'.format(lines, words, bytes, f.name))

tests/bin/wc.bat

@@ -0,0 +1,15 @@
+@echo on
+call :s_which py.exe
+rem note that %~dp0 is dir of this batch script
+if not "%_path%" == "" (
+  py -3 %~dp0wc %*
+) else (
+  python %~dp0wc %*
+)
+
+goto :eof
+
+:s_which
+  setlocal
+  endlocal & set _path=%~$PATH:1
+  goto :eof

tests/test_integrations.py

@@ -132,6 +132,13 @@ def _g(args, stdin=None):
 aliases['g'] = _g
 g
 """, (("x"*100) + '\n') * 1000, 0),
+# test piping 'real' command
+("""
+with open('tttt', 'w') as fp:
+    fp.write("Wow mom!\\n")
+
+![cat tttt | wc]
+""", '      1       2      10\n' if ON_WINDOWS else " 1  2 9 <stdin>\n", 0),
 ]
 
 
@@ -145,7 +152,7 @@ def test_script(case):
 
 @skip_if_on_windows
 @pytest.mark.parametrize('cmd, fmt, exp', [
-    ('pwd', None, os.getcwd() + '\n'),
+    ('pwd', None, lambda: os.getcwd() + '\n'),
     ('echo WORKING', None, 'WORKING\n'),
     ('ls -f', lambda out: out.splitlines().sort(), os.listdir().sort()),
     ])
@@ -156,17 +163,21 @@ def test_single_command(cmd, fmt, exp):
     out, err, rtn = run_xonsh(cmd, stderr=sp.DEVNULL)
     if callable(fmt):
         out = fmt(out)
+    if callable(exp):
+        exp = exp()
     assert out == exp
     assert rtn == 0
 
 
 @skip_if_on_windows
 @pytest.mark.parametrize('cmd, exp', [
-    ('pwd', os.getcwd() + '\n'),
+    ('pwd', lambda: os.getcwd() + '\n'),
     ])
 def test_redirect_out_to_file(cmd, exp, tmpdir):
     outfile = tmpdir.mkdir('xonsh_test_dir').join('xonsh_test_file')
     command = '{} > {}'.format(cmd, outfile)
     out, _, _ = run_xonsh(command)
     content = outfile.read()
+    if callable(exp):
+        exp = exp()
     assert content == exp

xonsh/proc.py

@@ -1484,6 +1484,7 @@ class CommandPipeline:
             stderr = NonBlockingFDReader(stderr.fileno(), timeout=timeout)
         # read from process while it is running
         check_prev_done = len(self.procs) == 1
+        prev_end_time = None
         while proc.poll() is None:
             if getattr(proc, 'suspended', False):
                 return
@@ -1504,8 +1505,22 @@ class CommandPipeline:
             yield from stdout_lines
             stderr_lines = safe_readlines(stderr, 1024)
             self.stream_stderr(stderr_lines)
-            if not check_prev_done and (stdout_lines or stderr_lines):
-                check_prev_done = True
+            if not check_prev_done:
+                # if we are piping...
+                if (stdout_lines or stderr_lines):
+                    # see if we have some output.
+                    check_prev_done = True
+                elif prev_end_time is None:
+                    # or see if we already know that the next-to-last
+                    # proc in teh pipeline has ended.
+                    if self.procs[-2].poll() is not None:
+                        # if it has, record the time
+                        prev_end_time = time.time()
+                elif time.time() - prev_end_time >= 0.1:
+                    # if we still don't have any output, even though the
+                    # next-to-last proc has finished, wait a bit to make
+                    # sure we have fully started up, etc.
+                    check_prev_done = True
             time.sleep(timeout)
         # read from process now that it is over
         yield from safe_readlines(stdout)