Class: OFlow::Actors::ShellOne
- Inherits:
-
OFlow::Actor
- Object
- OFlow::Actor
- OFlow::Actors::ShellOne
- Defined in:
- lib/oflow/actors/shellone.rb
Instance Attribute Summary collapse
-
#cmd ⇒ Object
readonly
Returns the value of attribute cmd.
-
#dir ⇒ Object
readonly
Returns the value of attribute dir.
-
#timeout ⇒ Object
readonly
Returns the value of attribute timeout.
Attributes inherited from OFlow::Actor
Instance Method Summary collapse
-
#initialize(task, options) ⇒ ShellOne
constructor
A new instance of ShellOne.
- #perform(op, box) ⇒ Object
Methods inherited from OFlow::Actor
#busy?, #inputs, #options, #outputs, #set_option, #with_own_thread
Constructor Details
#initialize(task, options) ⇒ ShellOne
Returns a new instance of ShellOne.
14 15 16 17 18 19 20 21 22 23 |
# File 'lib/oflow/actors/shellone.rb', line 14 def initialize(task, ) super @dir = [:dir] @dir = '.' if @dir.nil? @dir = File.(@dir.strip) @cmd = [:cmd] @timeout = .fetch(:timeout, 1.0).to_f @timeout = 0.001 if 0.001 > @timeout end |
Instance Attribute Details
#cmd ⇒ Object (readonly)
Returns the value of attribute cmd.
11 12 13 |
# File 'lib/oflow/actors/shellone.rb', line 11 def cmd @cmd end |
#dir ⇒ Object (readonly)
Returns the value of attribute dir.
10 11 12 |
# File 'lib/oflow/actors/shellone.rb', line 10 def dir @dir end |
#timeout ⇒ Object (readonly)
Returns the value of attribute timeout.
12 13 14 |
# File 'lib/oflow/actors/shellone.rb', line 12 def timeout @timeout end |
Instance Method Details
#perform(op, box) ⇒ Object
25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 |
# File 'lib/oflow/actors/shellone.rb', line 25 def perform(op, box) input = Oj.dump(box.contents, mode: :compat, indent: 0) i, o, e, _ = Open3.popen3(@cmd, chdir: @dir) i.write(input) i.close giveup = Time.now + @timeout ra = [e, o] out = '' err = '' ec = false # stderr closed flag oc = false # stdout closed flag while true rem = giveup - Time.now raise Exception.new("Timed out waiting for output.") if 0.0 > rem rs, _, es = select(ra, nil, ra, rem) unless es.nil? es.each do |io| ec |= io == e oc |= io == o end end break if ec && oc unless rs.nil? rs.each do |io| if io == e && !ec if io.closed? || io.eof? ec = true next end err += io.read_nonblock(1000) elsif io == o && !oc if io.closed? || io.eof? oc = true next end out += io.read_nonblock(1000) end end end break if ec && oc end if 0 < err.length raise Exception.new(err) end output = Oj.load(out, mode: :compat) o.close e.close task.ship(nil, Box.new(output, box.tracker)) end |