Class: RSpec::Conductor::Util::ChildProcess

Inherits:
Object
  • Object
show all
Defined in:
lib/rspec/conductor/util/child_process.rb

Constant Summary collapse

POLL_INTERVAL = 
0.01

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(on_stdout: nil, on_stderr: nil, on_message: nil) ⇒ ChildProcess

Returns a new instance of ChildProcess.



37
38
39
40
41
42
43
44
45
46
47
48
49
50
 
# File 'lib/rspec/conductor/util/child_process.rb', line 37

def initialize(on_stdout: nil, on_stderr: nil, on_message: nil)
  @on_stdout = on_stdout
  @on_stderr = on_stderr
  @on_message = on_message
  @pid = nil
  @exit_status = nil
  @stdout_pipe = nil
  @stderr_pipe = nil
  @stdout_buffer = +""
  @stderr_buffer = +""
  @message_socket = nil
  @ios = []
  @done = false
end

Instance Attribute Details

#exit_status ⇒ Object (readonly)

Returns the value of attribute exit_status.



11
12
13
 
# File 'lib/rspec/conductor/util/child_process.rb', line 11

def exit_status
  @exit_status
end

#ios ⇒ Object (readonly)

Returns the value of attribute ios.



11
12
13
 
# File 'lib/rspec/conductor/util/child_process.rb', line 11

def ios
  @ios
end

#message_socket ⇒ Object (readonly)

Returns the value of attribute message_socket.



11
12
13
 
# File 'lib/rspec/conductor/util/child_process.rb', line 11

def message_socket
  @message_socket
end

#pid ⇒ Object (readonly)

Returns the value of attribute pid.



11
12
13
 
# File 'lib/rspec/conductor/util/child_process.rb', line 11

def pid
  @pid
end

Class Method Details

.fork(**args, &block) ⇒ Object 



13
14
15
 
# File 'lib/rspec/conductor/util/child_process.rb', line 13

def self.fork(**args, &block)
  new(**args).fork(&block)
end

.tick_all(processes, poll_interval: POLL_INTERVAL) ⇒ Object 



25
26
27
28
29
30
31
32
33
34
35
 
# File 'lib/rspec/conductor/util/child_process.rb', line 25

def self.tick_all(processes, poll_interval: POLL_INTERVAL)
  processes_by_io = processes.each_with_object({}) do |process, memo|
    process.ios.reject(&:closed?).each { |io| memo[io] = process }
  end
  return false if processes_by_io.empty?

  ready, = IO.select(processes_by_io.keys, nil, nil, poll_interval)
  ready&.each { |io| processes_by_io[io].handle_available(io) }

  true
end

.wait_all(processes) ⇒ Object 



17
18
19
20
21
22
23
 
# File 'lib/rspec/conductor/util/child_process.rb', line 17

def self.wait_all(processes)
  until processes.all?(&:done?)
    break unless tick_all(processes)
  end

  processes.each(&:finalize)
end

Instance Method Details

#done? ⇒ Boolean

Returns:

  • (Boolean) 


97
98
99
 
# File 'lib/rspec/conductor/util/child_process.rb', line 97

def done?
  @done
end

#finalize ⇒ Object 



138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
 
# File 'lib/rspec/conductor/util/child_process.rb', line 138

def finalize
  return if done?

  @done = true

  process_buffer(@stdout_buffer, @on_stdout, drain_remaining: true)
  process_buffer(@stderr_buffer, @on_stderr, drain_remaining: true)

  begin
    _, status = Process.waitpid2(@pid)
    @exit_status = status.exitstatus
  rescue Errno::ECHILD
  end

  @message_socket&.close

  self
end

#fork(&block) ⇒ Object

Raises:

  • (ArgumentError) 


52
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
90
91
92
93
94
95
 
# File 'lib/rspec/conductor/util/child_process.rb', line 52

def fork(&block)
  raise ArgumentError, '.fork should be called with a block' unless block_given?

  stdout_read, stdout_write = IO.pipe
  stderr_read, stderr_write = IO.pipe
  parent_socket, child_socket = Socket.pair(:UNIX, :STREAM, 0) if @on_message

  @stdout_pipe = stdout_read
  @stderr_pipe = stderr_read
  @message_socket = parent_socket
  @ios = [@stdout_pipe, @stderr_pipe, @message_socket].compact

  @pid = Kernel.fork do
    stdout_read.close
    stderr_read.close
    parent_socket&.close

    $stdout = stdout_write
    $stderr = stderr_write
    $stdin = File.open("/dev/null")
    STDOUT.reopen($stdout)
    STDERR.reopen($stderr)
    STDIN.reopen($stdin)

    begin
      yield self, child_socket
    rescue => e
      stderr_write.puts "#{e.class}: #{e.message}\n#{e.backtrace.join("\n")}"
      exit 1
    ensure
      stdout_write.close
      stderr_write.close
      child_socket&.close
    end

    exit 0
  end

  stdout_write.close
  stderr_write.close
  child_socket&.close

  self
end

#handle_available(io) ⇒ Object 



101
102
103
104
105
106
107
108
109
110
111
 
# File 'lib/rspec/conductor/util/child_process.rb', line 101

def handle_available(io)
  return if done?
  return if io.closed?

  case io
  when @stdout_pipe, @stderr_pipe
    read_pipe(io)
  when @message_socket
    @on_message&.call
  end
end

#read_pipe(pipe) ⇒ Object 



113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
 
# File 'lib/rspec/conductor/util/child_process.rb', line 113

def read_pipe(pipe)
  buffer, callback = case pipe
                     when @stdout_pipe
                       [@stdout_buffer, @on_stdout]
                     when @stderr_pipe
                       [@stderr_buffer, @on_stderr]
                     else
                       return
                     end

  begin
    data = pipe.read_nonblock(4096, exception: false)
    if data == :wait_readable
      return
    elsif data.nil? || data.empty?
      pipe.close
    else
      buffer << data
      process_buffer(buffer, callback)
    end
  rescue IOError, EOFError
    pipe.close
  end
end

#success? ⇒ Boolean

Returns:

  • (Boolean) 


161
162
163
 
# File 'lib/rspec/conductor/util/child_process.rb', line 161

def success?
  @exit_status == 0
end

#wait ⇒ Object 



157
158
159
 
# File 'lib/rspec/conductor/util/child_process.rb', line 157

def wait
  self.class.wait_all([self])
end