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

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Receiver

Returns a new instance of Receiver.



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

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|
          if reader.eof?
            op = @readers[reader]
            @readers.delete_if { |key, value| key == reader }
            removed op[:process], op[:name]
          else
            op = @readers[reader]
            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.



10
11
12
 
# File 'lib/multi_process/receiver.rb', line 10

def mutex
  @mutex
end

Instance Method Details

#message(process, name, message) ⇒ Object

Send a custom messages.



53
54
55
 
# File 'lib/multi_process/receiver.rb', line 53

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`.



44
45
46
47
48
49
 
# File 'lib/multi_process/receiver.rb', line 44

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