From: Michal Suchanek <hramrach@...>
Date: 2011-05-16T20:05:40+09:00
Subject: Re: Shell pipeline in Ruby?

On 14 May 2011 17:05, Brian Candler <b.candler@pobox.com> wrote:
> 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

I don't use Ruby for that.

And I used a select loop because I had issues with threads in the past.

However, your example works perfectly for me, even when updated to put
more data through each of the pipes to make sure they don't end up all
in a buffer and the threads have to be actually switched for the
pipeline to work.

I think there might still be issues if the produced data was without
line endings because none of the gets would finish but that can be
solved by using read_nonblock or somesuch instead.

Thanks

Michal

-----------
#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('while read x ; do echo "$x" ; echo "$x" >&2 ; done')
    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('while read x ; do x=`echo "$x" | tr a-z A-Z` ; echo "$x" ;
echo "$x" >&2 ; done')
    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('while read x ; do x=`echo "$x" | sed -e s/O/0/g` ; echo "$x"
; echo "$x" >&2 ; done')
    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"
    (1..1000).each{|i|
        ruby_to_a_wr.puts "#{i} 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