https://gist.github.com/4600139
require "childprocess"
class Pipes
class ProcessBit
class Future
def initialize(process)
@process = process
return
end
def wait
@process.wait
return
end
end
def initialize(args)
@process = ChildProcess.build(*args)
return
end
def get_o
return @process.io.stdin
end
def start(o)
@process.duplex = true
@process.io.stdout = o
@process.start
o.close
return Future.new(@process)
end
end
class BlockBit
class Future
def initialize(thread)
@thread = thread
return
end
def wait
@thread.join
return
end
end
def initialize(block)
@block = block
@i, @o = IO.pipe
@i.close_on_exec = true
@o.close_on_exec = true
return
end
def get_o
return @o
end
def start(o)
thread = Thread.new do
begin
@block.call(@i, o)
ensure
@i.close
o.close
end
end
return Future.new(thread)
end
end
def initialize
@bits = []
return
end
def to(*args, &block)
if block_given?
@bits.push(BlockBit.new(block))
else
@bits.push(ProcessBit.new(args))
end
return self
end
def run
futures = []
o = File.open("/dev/null", mode: IO::WRONLY)
@bits.reverse_each do |bit|
futures.push(bit.start(o))
o = bit.get_o
end
o.close
futures.reverse_each do |future|
future.wait
end
return
end
end
def tee(s, i, o)
i.each_line do |line|
STDOUT.write("#{s}:#{line}")
o.write(line)
end
return
end
Pipes.new.to("ls").to("sort").to { |i, o|
tee("1", i, o)
}.to("sort", "-r").to { |i, o|
tee("2", i, o)
}.run
Thanks,
Nicholas
--
You received this message because you are subscribed to the Google Groups "Ruby or
Rails Oceania" group.
To post to this group, send email to [email protected].
To unsubscribe from this group, send email to
[email protected].
For more options, visit this group at
http://groups.google.com/group/rails-oceania?hl=en.
