Re: Shell pipeline in Ruby?

Michal Suchanek wrote in post #998548:
> And the long script with pipes that calls the above:

Since you are doing this on a real operating system, there's no need to 
use ruby to copy the output of one process to the input of the next 
process (unless you want to capture the communication between the 
processes as well).

OTOH, making sure all the right FDs are closed in each child can be 
tricky in 1.8, because there's no IO.close_on_exec= (although 1.9 has 
this).

Here is an example which works in 1.8.7. Obviously it's possible to 
refactor this to N children, although I expect you'll end up with 
something like the 'pipeline' method mentioned before if you do.

--------------------------------
#Thread.abort_on_exception = true  # for debugging

ruby_to_a_rd, ruby_to_a_wr = IO.pipe
a_to_b_rd, a_to_b_wr = IO.pipe
a_err_rd, a_err_wr = IO.pipe
# open: r2a_r, r2a_w, a2b_r, a2b_w, ae_r, ae_w
pid1 = fork do
  ruby_to_a_wr.close
  a_to_b_rd.close
  a_err_rd.close
  STDIN.reopen(ruby_to_a_rd)
  STDOUT.reopen(a_to_b_wr)
  STDERR.reopen(a_err_wr)
  exec("cat; echo done cat 1>&2")
  STDERR.puts "Whoops! #{$!}"
end
ruby_to_a_rd.close
a_to_b_wr.close
a_err_wr.close
# open: r2a_w, a2b_r, ae_r

b_to_c_rd, b_to_c_wr = IO.pipe
b_err_rd, b_err_wr = IO.pipe
# open: r2a_w, a2b_r, ae_r, b2c_r, b2c_w, be_r, be_w
pid2 = fork do
  ruby_to_a_wr.close
  b_to_c_rd.close
  a_err_rd.close
  b_err_rd.close
  STDIN.reopen(a_to_b_rd)
  STDOUT.reopen(b_to_c_wr)
  STDERR.reopen(b_err_wr)
  exec("tr 'a-z' 'A-Z'; echo done tr 1>&2")
  STDERR.puts "Whoops! #{$!}"
end
a_to_b_rd.close
b_to_c_wr.close
b_err_wr.close
# open: r2a_w, ae_r, b2c_r, be_r

c_to_ruby_rd, c_to_ruby_wr = IO.pipe
c_err_rd, c_err_wr = IO.pipe
# open: r2a_w, ae_r, b2c_r, be_r, c2r_r, c2r_w, ce_r, ce_w
pid3 = fork do
  ruby_to_a_wr.close
  c_to_ruby_rd.close
  a_err_rd.close
  b_err_rd.close
  c_err_rd.close
  STDIN.reopen(b_to_c_rd)
  STDOUT.reopen(c_to_ruby_wr)
  STDERR.reopen(c_err_wr)
  exec("sed 's/O/0/g'; echo done sed 1>&2")
  STDERR.puts "Whoops! #{$!}"
end
b_to_c_rd.close
c_to_ruby_wr.close
c_err_wr.close
# open: r2a_w, ae_r, be_r, c2r_r, ce_r

Thread.new do
  ruby_to_a_wr.puts "Here is some data"
  ruby_to_a_wr.puts "And some more"
  ruby_to_a_wr.close
end

Thread.new do
  while line = a_err_rd.gets
    puts "A err: #{line}"
  end
  a_err_rd.close
end

Thread.new do
  while line = b_err_rd.gets
    puts "B err: #{line}"
  end
  b_err_rd.close
end

Thread.new do
  while line = c_err_rd.gets
    puts "C err: #{line}"
  end
  c_err_rd.close
end

while line = c_to_ruby_rd.gets
  puts line
end
c_to_ruby_rd.close
--------------------------------

This can be simplified a bit if you don't want three separate error 
streams; they can all share the same one. i.e. each child would have

  err_rd.close
  STDERR.reopen(err_wr)

Regards,

Brian.

-- 
Posted via http://www.ruby-forum.com/.