* lib/open3.rb (Open3.popen3): simplified.

(Open3.popen_run): extracted from Open3.popen3.
  (Open3.popen2): new method.
  (Open3.popen2e): new method.
  (Open3.pipeline_rw): new method.
  (Open3.pipeline_r): new method.
  (Open3.pipeline_w): new method.
  (Open3.pipeline_run): new private method.



git-svn-id: svn+ssh://ci.ruby-lang.org/ruby/trunk@20512 b2dd03c8-39d4-4d8f-98ff-823fe69b080e

ChangeLog

@@ -1,3 +1,14 @@
+Thu Dec  4 19:56:20 2008  Tanaka Akira  <akr@fsij.org>
+
+	* lib/open3.rb (Open3.popen3): simplified.
+	  (Open3.popen_run): extracted from Open3.popen3.
+	  (Open3.popen2): new method.
+	  (Open3.popen2e): new method.
+	  (Open3.pipeline_rw): new method.
+	  (Open3.pipeline_r): new method.
+	  (Open3.pipeline_w): new method.
+	  (Open3.pipeline_run): new private method.
+
 Thu Dec  4 19:16:28 2008  Tanaka Akira  <akr@fsij.org>
 
 	* process.c (check_exec_fds): resolve cascaded child fd reference.

lib/open3.rb

@@ -51,7 +51,7 @@ module Open3
   #   stdin, stdout, stderr, wait_thr = Open3.popen3(cmd... [, opts])
   #   pid = wait_thr[:pid]  # pid of the started process.
   #   ...
-  #   stdin.close  # stdin, stdout and stderr should be closed in this form.
+  #   stdin.close  # stdin, stdout and stderr should be closed explicitly in this form.
   #   stdout.close
   #   stderr.close
   #   exit_status = wait_thr.value  # Process::Status object returned.
@@ -61,6 +61,7 @@ module Open3
   #
   #   Open3.popen3("echo a") {|i, o, e, t| ... }
   #   Open3.popen3("echo", "a") {|i, o, e, t| ... }
+  #   Open3.popen3(["echo", "argv0"], "a") {|i, o, e, t| ... }
   #
   # If the last parameter, opts, is a Hash, it is recognized as an option for Kernel#spawn.
   #
@@ -68,75 +69,105 @@ module Open3
   #     p o.read.chomp #=> "/"
   #   }
   #
-  # opts[STDIN], opts[STDOUT] and opts[STDERR] in the option are set for redirection.
-  #
-  # If some of the three elements in opts are specified,
-  # pipes for them are not created.
-  # In that case, block arugments for the block form and
-  # return values for the non-block form are decreased.
-  #
-  #   # No pipe "e" for stderr
-  #   Open3.popen3("echo a", STDERR=>nil) {|i,o,t| ... }
-  #   i,o,t = Open3.popen3("echo a", STDERR=>nil)
-  #
-  # If the value is nil as above, the elements of opts are removed.
-  # So standard input/output/error of current process are inherited.
-  #
-  # If the value is not nil, it is passed as is to Kernel#spawn.
-  # So pipeline of commands can be constracted as follows.
-  #
-  #   Open3.popen3("yes", STDIN=>nil, STDERR=>nil) {|o1,t1|
-  #     Open3.popen3("head -10", STDIN=>o1, STDERR=>nil) {|o2,t2|
-  #       o1.close
-  #       p o2.read     #=> "y\ny\ny\ny\ny\ny\ny\ny\ny\ny\n"
-  #       p t1.value    #=> #<Process::Status: pid 13508 SIGPIPE (signal 13)>
-  #       p t2.value    #=> #<Process::Status: pid 13510 exit 0>
-  #     } 
-  #   }
-  #
   # wait_thr.value waits the termination of the process.
   # The block form also waits the process when it returns.
   #
   # Closing stdin, stdout and stderr does not wait the process.
   #
-  def popen3(*cmd)
+  def popen3(*cmd, &block)
     if Hash === cmd.last
       opts = cmd.pop.dup
     else
       opts = {}
     end
 
-    child_io = []
+    in_r, in_w = IO.pipe
-    parent_io = []
+    opts[STDIN] = in_r
+    in_w.sync = true
 
-    if !opts.include?(STDIN)
+    out_r, out_w = IO.pipe
-      pw = IO.pipe   # pipe[0] for read, pipe[1] for write
+    opts[STDOUT] = out_w
-      opts[STDIN] = pw[0]
+
-      pw[1].sync = true
+    err_r, err_w = IO.pipe
-      child_io << pw[0]
+    opts[STDERR] = err_w
-      parent_io << pw[1]
+
-    elsif opts[STDIN] == nil
+    popen_run(cmd, opts, [in_r, out_w, err_w], [in_w, out_r, err_r], &block)
-      opts.delete(STDIN)
+  end
+  module_function :popen3
+
+  # Open3.popen2 is similer to Open3.popen3 except it doesn't make a pipe for
+  # the standard error stream.
+  #
+  # Block form:
+  #
+  #   Open3.popen2(cmd... [, opts]) {|stdin, stdout, wait_thr|
+  #     pid = wait_thr.pid # pid of the started process.
+  #     ...
+  #     exit_status = wait_thr.value # Process::Status object returned.
+  #   }
+  #
+  # Non-block form:
+  #   
+  #   stdin, stdout, wait_thr = Open3.popen2(cmd... [, opts])
+  #   ...
+  #   stdin.close  # stdin and stdout should be closed explicitly in this form.
+  #   stdout.close
+  #
+  def popen2(*cmd, &block)
+    if Hash === cmd.last
+      opts = cmd.pop.dup
+    else
+      opts = {}
     end
 
-    if !opts.include?(STDOUT)
+    in_r, in_w = IO.pipe
-      pr = IO.pipe
+    opts[STDIN] = in_r
-      opts[STDOUT] = pr[1]
+    in_w.sync = true
-      child_io << pr[1]
+
-      parent_io << pr[0]
+    out_r, out_w = IO.pipe
-    elsif opts[STDOUT] == nil
+    opts[STDOUT] = out_w
-      opts.delete(STDOUT)
+
+    popen_run(cmd, opts, [in_r, out_w], [in_w, out_r], &block)
+  end
+  module_function :popen2
+
+  # Open3.popen2e is similer to Open3.popen3 except it merges
+  # the standard output stream and the standard error stream.
+  #
+  # Block form:
+  #
+  #   Open3.popen2e(cmd... [, opts]) {|stdin, stdout_and_stderr, wait_thr|
+  #     pid = wait_thr.pid # pid of the started process.
+  #     ...
+  #     exit_status = wait_thr.value # Process::Status object returned.
+  #   }
+  #
+  # Non-block form:
+  #   
+  #   stdin, stdout_and_stderr, wait_thr = Open3.popen2e(cmd... [, opts])
+  #   ...
+  #   stdin.close  # stdin and stdout_and_stderr should be closed explicitly in this form.
+  #   stdout_and_stderr.close
+  #
+  def popen2e(*cmd, &block)
+    if Hash === cmd.last
+      opts = cmd.pop.dup
+    else
+      opts = {}
     end
 
-    if !opts.include?(STDERR)
+    in_r, in_w = IO.pipe
-      pe = IO.pipe
+    opts[STDIN] = in_r
-      opts[STDERR] = pe[1]
+    in_w.sync = true
-      child_io << pe[1]
-      parent_io << pe[0]
-    elsif opts[STDERR] == nil
-      opts.delete(STDERR)
-    end
 
+    out_r, out_w = IO.pipe
+    opts[[STDOUT, STDERR]] = out_w
+
+    popen_run(cmd, opts, [in_r, out_w], [in_w, out_r], &block)
+  end
+  module_function :popen2e
+
+  def popen_run(cmd, opts, child_io, parent_io) # :nodoc:
     pid = spawn(*cmd, opts)
     wait_thr = Process.detach(pid)
     child_io.each {|io| io.close }
@@ -151,7 +182,185 @@ module Open3
     end
     result
   end
-  module_function :popen3
+  module_function :popen_run
+  class << self
+    private :popen_run
+  end
+
+  # Open3.pipeline_rw starts list of commands as a pipeline with pipes
+  # which connects stdin of the first command and stdout of the last command.
+  #
+  #   Open3.pipeline_rw(cmd1, cmd2, ... [, opts]) {|stdin, stdout, wait_threads|
+  #     ...
+  #   }
+  #
+  #   stdin, stdout, wait_threads = Open3.pipeline_rw(cmd1, cmd2, ... [, opts])
+  #   ...
+  #   stdin.close
+  #   stdout.close
+  #
+  # Each cmd is a string or an array.
+  # If it is an array, the elements are passed to Kernel#spawn.
+  #
+  # The option to pass Kernel#spawn is constructed by merging
+  # +opts+, the last hash element of the array and
+  # specification for the pipe between each commands.
+  #
+  # Example:
+  #
+  #   Open3.pipeline_rw("sort", "cat -n") {|stdin, stdout, wait_thrs|
+  #     stdin.puts "foo"
+  #     stdin.puts "bar"
+  #     stdin.puts "baz"
+  #     stdin.close     # send EOF to sort.
+  #     p stdout.read   #=> "     1\tbar\n     2\tbaz\n     3\tfoo\n"
+  #   }
+  def pipeline_rw(*cmds, &block)
+    if Hash === cmds.last
+      opts = cmds.pop.dup
+    else
+      opts = {}
+    end
+
+    in_r, in_w = IO.pipe
+    opts[STDIN] = in_r
+    in_w.sync = true
+
+    out_r, out_w = IO.pipe
+    opts[STDOUT] = out_w
+
+    pipeline_run(cmds, opts, [in_r, out_w], [in_w, out_r], &block)
+  end
+  module_function :pipeline_rw
+
+  # Open3.pipeline_r starts list of commands as a pipeline with a pipe
+  # which connects stdout of the last command.
+  #
+  #   Open3.pipeline_r(cmd1, cmd2, ... [, opts]) {|stdout, wait_threads|
+  #     ...
+  #   }
+  #
+  #   stdout, wait_threads = Open3.pipeline_r(cmd1, cmd2, ... [, opts])
+  #   ...
+  #   stdout.close
+  #
+  # Example:
+  #
+  #   Open3.pipeline_r("yes", "head -10") {|r, ts|
+  #     p r.read      #=> "y\ny\ny\ny\ny\ny\ny\ny\ny\ny\n"
+  #     p ts[0].value #=> #<Process::Status: pid 24910 SIGPIPE (signal 13)>
+  #     p ts[1].value #=> #<Process::Status: pid 24913 exit 0>
+  #   }
+  #
+  def pipeline_r(*cmds, &block)
+    if Hash === cmds.last
+      opts = cmds.pop.dup
+    else
+      opts = {}
+    end
+
+    out_r, out_w = IO.pipe
+    opts[STDOUT] = out_w
+
+    pipeline_run(cmds, opts, [out_w], [out_r], &block)
+  end
+  module_function :pipeline_r
+
+  # Open3.pipeline_w starts list of commands as a pipeline with a pipe
+  # which connects stdin of the first command.
+  #
+  #   Open3.pipeline_w(cmd1, cmd2, ... [, opts]) {|stdin, wait_threads|
+  #     ...
+  #   }
+  #
+  #   stdin, wait_threads = Open3.pipeline_w(cmd1, cmd2, ... [, opts])
+  #   ...
+  #   stdin.close
+  #
+  # Example:
+  #
+  #   Open3.pipeline_w("cat -n", "bzip2 -c", STDOUT=>"/tmp/z.bz2") {|w, ts|
+  #     w.puts "hello" 
+  #     w.close
+  #     p ts[0].value
+  #     p ts[1].value
+  #   }
+  #
+  def pipeline_w(*cmds, &block)
+    if Hash === cmds.last
+      opts = cmds.pop.dup
+    else
+      opts = {}
+    end
+
+    in_r, in_w = IO.pipe
+    opts[STDIN] = in_r
+    in_w.sync = true
+
+    pipeline_run(cmds, opts, [in_r], [in_w], &block)
+  end
+  module_function :pipeline_w
+
+  def pipeline_run(cmds, pipeline_opts, child_io, parent_io, &block) # :nodoc:
+    if cmds.empty?
+      raise ArgumentError, "no commands"
+    end
+
+    opts_base = pipeline_opts.dup
+    opts_base.delete STDIN
+    opts_base.delete STDOUT
+
+    wait_thrs = []
+    r = nil
+    cmds.each_with_index {|cmd, i|
+      cmd_opts = opts_base.dup
+      if String === cmd
+        cmd = [cmd]
+      else
+        cmd_opts.update cmd.pop if Hash === cmd.last
+      end
+      if i == 0
+        if !cmd_opts.include?(STDIN)
+          if pipeline_opts.include?(STDIN)
+            cmd_opts[STDIN] = pipeline_opts[STDIN]
+          end
+        end
+      else
+        cmd_opts[STDIN] = r
+      end
+      if i != cmds.length - 1
+        r2, w2 = IO.pipe
+        cmd_opts[STDOUT] = w2
+      else
+        if !cmd_opts.include?(STDOUT)
+          if pipeline_opts.include?(STDOUT)
+            cmd_opts[STDOUT] = pipeline_opts[STDOUT]
+          end
+        end
+      end
+      pid = spawn(*cmd, cmd_opts)
+      wait_thrs << Process.detach(pid)
+      r.close if r
+      w2.close if w2
+      r = r2
+    }
+    result = parent_io + [wait_thrs]
+    child_io.each {|io| io.close }
+    if defined? yield
+      begin
+	return yield(*result)
+      ensure
+	parent_io.each{|io| io.close unless io.closed?}
+        wait_thrs.each {|t| t.join }
+      end
+    end
+    result
+  end
+  module_function :pipeline_run
+  class << self
+    private :pipeline_run
+  end
+
 end
 
 if $0 == __FILE__

test/test_open3.rb

@@ -74,21 +74,6 @@ class TestOpen3 < Test::Unit::TestCase
     }
   end
 
-  def test_disable
-    Open3.popen3(RUBY, '-e', '', STDIN=>nil) {|o,e,t|
-      assert_kind_of(Thread, t)
-    }
-    Open3.popen3(RUBY, '-e', '', STDOUT=>nil) {|i,e,t|
-      assert_kind_of(Thread, t)
-    }
-    Open3.popen3(RUBY, '-e', '', STDERR=>nil) {|i,o,t|
-      assert_kind_of(Thread, t)
-    }
-    Open3.popen3(RUBY, '-e', '', STDIN=>nil, STDOUT=>nil, STDERR=>nil) {|t|
-      assert_kind_of(Thread, t)
-    }
-  end
-
   def with_pipe
     r, w = IO.pipe
     yield r, w
@@ -106,100 +91,83 @@ class TestOpen3 < Test::Unit::TestCase
     old.close if old && !old.closed?
   end
 
-  def test_disable_stdin
+  def test_popen2
-    with_pipe {|r, w|
-      with_reopen(STDIN, r) {|old|
-        Open3.popen3(RUBY, '-e', 's=STDIN.read; STDOUT.print s+"o"; STDERR.print s+"e"', STDIN=>nil) {|o,e,t|
-          assert_kind_of(Thread, t)
-          w.print "x"
-          w.close
-          assert_equal("xo", o.read)
-          assert_equal("xe", e.read)
-        }
-      }
-    }
-  end
-
-  def test_disable_stdout
-    with_pipe {|r, w|
-      with_reopen(STDOUT, w) {|old|
-        w.close
-        Open3.popen3(RUBY, '-e', 's=STDIN.read; STDOUT.print s+"o"; STDERR.print s+"e"', STDOUT=>nil) {|i,e,t|
-          assert_kind_of(Thread, t)
-          i.print "y"
-          i.close
-          STDOUT.reopen(old)
-          assert_equal("yo", r.read)
-          assert_equal("ye", e.read)
-        }
-      }
-    }
-  end
-
-  def test_disable_stderr
     with_pipe {|r, w|
       with_reopen(STDERR, w) {|old|
         w.close
-        Open3.popen3(RUBY, '-e', 's=STDIN.read; STDOUT.print s+"o"; STDERR.print s+"e"', STDERR=>nil) {|i,o,t|
+        Open3.popen2(RUBY, '-e', 's=STDIN.read; STDOUT.print s+"o"; STDERR.print s+"e"') {|i,o,t|
+          assert_kind_of(Thread, t)
+          i.print "z"
+          i.close
+          STDERR.reopen(old)
+          assert_equal("zo", o.read)
+          assert_equal("ze", r.read)
+        }
+      }
+    }
+  end
+
+  def test_popen2e
+    with_pipe {|r, w|
+      with_reopen(STDERR, w) {|old|
+        w.close
+        Open3.popen2e(RUBY, '-e', 's=STDIN.read; STDOUT.print s+"o"; STDOUT.flush; STDERR.print s+"e"') {|i,o,t|
           assert_kind_of(Thread, t)
           i.print "y"
           i.close
           STDERR.reopen(old)
-          assert_equal("yo", o.read)
+          assert_equal("yoye", o.read)
-          assert_equal("ye", r.read)
+          assert_equal("", r.read)
         }
       }
     }
   end
 
-  def test_plug_pipe
+  def test_pipeline_rw
-    Open3.popen3(RUBY, '-e', 'STDOUT.print "1"') {|i1,o1,e1,t1|
+    Open3.pipeline_rw([RUBY, '-e', 'print STDIN.read + "1"'],
-      Open3.popen3(RUBY, '-e', 'STDOUT.print STDIN.read+"2"', STDIN=>o1) {|o2,e2,t2|
+                      [RUBY, '-e', 'print STDIN.read + "2"']) {|i,o,ts|
-        assert_equal("12", o2.read)
+      assert_kind_of(IO, i)
+      assert_kind_of(IO, o)
+      assert_kind_of(Array, ts)
+      assert_equal(2, ts.length)
+      ts.each {|t| assert_kind_of(Thread, t) }
+      i.print "0"
+      i.close
+      assert_equal("012", o.read)
+      ts.each {|t|
+        assert(t.value.success?)
       }
     }
   end
 
-  def test_tree_pipe
+  def test_pipeline_r
-    ia,oa,ea,ta = Open3.popen3(RUBY, '-e', 'i=STDIN.read; STDOUT.print i+"a"; STDERR.print i+"A"')
+    Open3.pipeline_r([RUBY, '-e', 'print "1"'],
-    ob,eb,tb = Open3.popen3(RUBY, '-e', 'i=STDIN.read; STDOUT.print i+"b"; STDERR.print i+"B"', STDIN=>oa)
+                     [RUBY, '-e', 'print STDIN.read + "2"']) {|o,ts|
-    oc,ec,tc = Open3.popen3(RUBY, '-e', 'i=STDIN.read; STDOUT.print i+"c"; STDERR.print i+"C"', STDIN=>ob)
+      assert_kind_of(IO, o)
-    od,ed,td = Open3.popen3(RUBY, '-e', 'i=STDIN.read; STDOUT.print i+"d"; STDERR.print i+"D"', STDIN=>eb)
+      assert_kind_of(Array, ts)
-    oe,ee,te = Open3.popen3(RUBY, '-e', 'i=STDIN.read; STDOUT.print i+"e"; STDERR.print i+"E"', STDIN=>ea)
+      assert_equal(2, ts.length)
-    of,ef,tf = Open3.popen3(RUBY, '-e', 'i=STDIN.read; STDOUT.print i+"f"; STDERR.print i+"F"', STDIN=>oe)
+      ts.each {|t| assert_kind_of(Thread, t) }
-    og,eg,tg = Open3.popen3(RUBY, '-e', 'i=STDIN.read; STDOUT.print i+"g"; STDERR.print i+"G"', STDIN=>ee)
+      assert_equal("12", o.read)
-    oa.close
+      ts.each {|t|
-    ea.close
+        assert(t.value.success?)
-    ob.close
+      }
-    eb.close
+    }
-    oe.close
-    ee.close
-
-    ia.print "0"
-    ia.close
-    assert_equal("0abc", oc.read)
-    assert_equal("0abC", ec.read)
-    assert_equal("0aBd", od.read)
-    assert_equal("0aBD", ed.read)
-    assert_equal("0Aef", of.read)
-    assert_equal("0AeF", ef.read)
-    assert_equal("0AEg", og.read)
-    assert_equal("0AEG", eg.read)
-  ensure
-    ia.close if !ia.closed?
-    oa.close if !oa.closed?
-    ea.close if !ea.closed?
-    ob.close if !ob.closed?
-    eb.close if !eb.closed?
-    oc.close if !oc.closed?
-    ec.close if !ec.closed?
-    od.close if !od.closed?
-    ed.close if !ed.closed?
-    oe.close if !oe.closed?
-    ee.close if !ee.closed?
-    of.close if !of.closed?
-    ef.close if !ef.closed?
-    og.close if !og.closed?
-    eg.close if !eg.closed?
   end
+
+  def test_pipeline_w
+    command = [RUBY, '-e', 's=STDIN.read; print s[1..-1]; exit s[0] == ?t']
+    str = 'ttftff'
+    Open3.pipeline_w(*[command]*str.length) {|i,ts|
+      assert_kind_of(IO, i)
+      assert_kind_of(Array, ts)
+      assert_equal(str.length, ts.length)
+      ts.each {|t| assert_kind_of(Thread, t) }
+      i.print str
+      i.close
+      ts.each_with_index {|t, i|
+        assert_equal(str[i] == ?t, t.value.success?)
+      }
+    }
+  end
+
 end