Make run() fully non-blocking and fix potential other problems

src/Symfony/Component/Process/Process.php

@@ -91,16 +91,16 @@ class Process
         $this->stdout = '';
         $this->stderr = '';
         $that = $this;
-        $callback = function ($type, $line) use ($that, $callback)
+        $callback = function ($type, $data) use ($that, $callback)
         {
             if ('out' == $type) {
-                $that->addOutput($line);
+                $that->addOutput($data);
             } else {
-                $that->addErrorOutput($line);
+                $that->addErrorOutput($data);
             }
 
             if (null !== $callback) {
-                call_user_func($callback, $type, $line);
+                call_user_func($callback, $type, $data);
             }
         };
 
@@ -115,21 +115,27 @@ class Process
 
         $process = proc_open($this->commandline, $descriptors, $pipes, $this->cwd, $this->env, $this->options);
 
-        stream_set_blocking($pipes[1], false);
-        stream_set_blocking($pipes[2], false);
-
         if (!is_resource($process)) {
             throw new \RuntimeException('Unable to launch a new process.');
         }
 
-        if (null !== $this->stdin) {
-            fwrite($pipes[0], (binary) $this->stdin);
+        foreach ($pipes as $pipe) {
+            stream_set_blocking($pipe, false);
+        }
+        
+        if (null === $this->stdin) {
+            fclose($pipes[0]);
+            $writePipes = null; 
+        } else {
+            $writePipes = array($pipes[0]);
+            $stdinLen = strlen($this->stdin);
+            $stdinOffset = 0;
         }
-        fclose($pipes[0]);
+        unset($pipes[0]);
 
-        while (true) {
+        while ($pipes || $writePipes) {
             $r = $pipes;
-            $w = null;
+            $w = $writePipes;
             $e = null;
 
             $n = @stream_select($r, $w, $e, $this->timeout);
@@ -140,28 +146,28 @@ class Process
                 proc_terminate($process);
 
                 throw new \RuntimeException('The process timed out.');
-            } elseif ($n > 0) {
-                $called = false;
-
-                while (true) {
-                    $c = false;
-                    if ($line = (binary) fgets($pipes[1], 1024)) {
-                        $called = $c = true;
-                        call_user_func($callback, 'out', $line);
-                    }
-
-                    if ($line = fgets($pipes[2], 1024)) {
-                        $called = $c = true;
-                        call_user_func($callback, 'err', $line);
-                    }
-
-                    if (!$c) {
-                        break;
-                    }
+            }
+
+            if ($w) {
+                $written = fwrite($writePipes[0], (binary) substr($this->stdin, $stdinOffset), 8192);
+                if (false !== $written) {
+                    $stdinOffset += $written;
+                }
+                if ($stdinOffset >= $stdinLen) {
+                    fclose($writePipes[0]);
+                    $writePipes = null;
                 }
+            }
 
-                if (!$called) {
-                    break;
+            foreach ($r as $pipe) {
+                $type = array_search($pipe, $pipes);
+                $data = fread($pipe, 8192);
+                if (strlen($data) > 0) {
+                    call_user_func($callback, $type == 1 ? 'out' : 'err', $data);
+                }
+                if (false === $data || feof($pipe)) {
+                    fclose($pipe);
+                    unset($pipes[$type]);
                 }
             }
         }