Class: Magritte::Pipe
- Inherits:
-
Object
- Object
- Magritte::Pipe
- Defined in:
- lib/magritte/pipe.rb
Class Method Summary collapse
Instance Method Summary collapse
- #filtering_with(command) ⇒ Object
-
#initialize(input, output = nil) ⇒ Pipe
constructor
A new instance of Pipe.
- #line_by_line ⇒ Object
- #out_to(io = nil, &block) ⇒ Object
- #separated_by(record_separator) ⇒ Object
Constructor Details
#initialize(input, output = nil) ⇒ Pipe
Returns a new instance of Pipe.
11 12 13 14 15 |
# File 'lib/magritte/pipe.rb', line 11 def initialize(input, output=nil) @input = input @output = output @line_by_line = false end |
Class Method Details
.from_input_file(infile) ⇒ Object
17 18 19 |
# File 'lib/magritte/pipe.rb', line 17 def self.from_input_file(infile) self.from_input_stream(File.open(File.(infile))) end |
.from_input_stream(io) ⇒ Object
21 22 23 |
# File 'lib/magritte/pipe.rb', line 21 def self.from_input_stream(io) new(BlockStream.new(io)) end |
.from_input_string(str) ⇒ Object
25 26 27 |
# File 'lib/magritte/pipe.rb', line 25 def self.from_input_string(str) self.from_input_stream((StringIO.new(str || ""))) end |
Instance Method Details
#filtering_with(command) ⇒ Object
53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 |
# File 'lib/magritte/pipe.rb', line 53 def filtering_with(command) raise "No output IO is set! Invoke out_to first!" unless @output Open3.popen2e(command) do |subproc_input, subproc_output, wait_thr| @subproc_output = BlockStream.new(subproc_output) @subproc_input = BlockStream.new(subproc_input) clear_buffers while true do read_ready, write_ready, = select read_from_input if write_ready write_to_subproc end if read_ready read_from_subproc write_output end # We close the input to signal to the sub-process that we are # done sending data. It will close its output when processsing # is completed. That signals us to stop piping data. if ready_to_close? @subproc_input.flush @subproc_input.close end break if @subproc_output.closed? end raise Errno::EPIPE.new("sub-process dirty exit!") unless wait_thr.value == 0 end end |
#line_by_line ⇒ Object
48 49 50 51 |
# File 'lib/magritte/pipe.rb', line 48 def line_by_line @line_by_line = true self end |
#out_to(io = nil, &block) ⇒ Object
29 30 31 32 33 34 35 36 37 38 39 40 41 42 |
# File 'lib/magritte/pipe.rb', line 29 def out_to(io=nil, &block) unless block_given? @output = BlockStream.new(io) return self end if @line_by_line @output = LineBufferOutputStream.new(block, @record_separator || "\n") else @output = ProcOutputStream.new(block) end self end |
#separated_by(record_separator) ⇒ Object
44 45 46 |
# File 'lib/magritte/pipe.rb', line 44 def separated_by(record_separator) @record_separator = record_separator end |