Class: MultiProcess::Receiver

Inherits:
Object
  • Object
show all
Defined in:
lib/multi_process/receiver.rb

Overview

Can handle input from multiple processes and run custom actions on event and output.

Direct Known Subclasses

Logger, NilReceiver, StringReceiver

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Receiver

Returns a new instance of Receiver.



10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
 
# File 'lib/multi_process/receiver.rb', line 10

def initialize
  @mutex   = Mutex.new
  @readers = {}

  Thread.new do
    begin
      loop do
        io = IO.select(@readers.keys, nil, nil, 0.1)
        (io.nil? ? [] : io.first).each do |reader|
          op = @readers[reader]

          if reader.eof?
            @readers.delete_if { |key, _value| key == reader }
            removed op[:process], op[:name]
          else
            received op[:process], op[:name], read(reader)
          end
        end
      end
    rescue Exception => ex
      puts ex.message
      puts ex.backtrace
    end
  end
end

Instance Attribute Details

#mutex ⇒ Object (readonly)

Mutex to synchronize operations.



8
9
10
 
# File 'lib/multi_process/receiver.rb', line 8

def mutex
  @mutex
end

Instance Method Details

#message(process, name, message) ⇒ Object

Send a custom messages.



51
52
53
 
# File 'lib/multi_process/receiver.rb', line 51

def message(process, name, message)
  received process, name, message
end

#pipe(process, name) ⇒ Object

Request a new pipe writer for given process and name.

Parameters:

  • process (Process) —

    Process requesting pipe.

  • name (Symbol) —

    Name associated to pipe e.g. ‘:out` or `:err`.



42
43
44
45
46
47
 
# File 'lib/multi_process/receiver.rb', line 42

def pipe(process, name)
  reader, writer = IO.pipe
  @readers[reader] = { name: name, process: process }
  connected process, name
  writer
end