Class: Rodimus::Transformation

Inherits:
Object
  • Object
show all
Includes:
Observable, Observing, RuntimeLogging
Defined in:
lib/rodimus/transformation.rb

Instance Attribute Summary collapse

Attributes included from RuntimeLogging

#start_time

Instance Method Summary collapse

Methods included from RuntimeLogging

#after_run_record_time, #before_run_record_time

Methods included from Observing

#on_notify

Methods included from Observable

#notify, #observers

Constructor Details

#initialize ⇒ Transformation

Returns a new instance of Transformation.



18
19
20
21
22
23
 
# File 'lib/rodimus/transformation.rb', line 18

def initialize
  @steps = []
  @ids = []
  @shared_data = {} # TODO: This needs to be thread safe
  observers << self
end

Instance Attribute Details

#drb_server ⇒ Object (readonly)

Returns the value of attribute drb_server.



10
11
12
 
# File 'lib/rodimus/transformation.rb', line 10

def drb_server
  @drb_server
end

#ids ⇒ Object (readonly)

Contains the thread or process identifiers currently in use



13
14
15
 
# File 'lib/rodimus/transformation.rb', line 13

def ids
  @ids
end

#shared_data ⇒ Object (readonly)

User-data accessible across all running steps.



16
17
18
 
# File 'lib/rodimus/transformation.rb', line 16

def shared_data
  @shared_data
end

#steps ⇒ Object (readonly)

Returns the value of attribute steps.



10
11
12
 
# File 'lib/rodimus/transformation.rb', line 10

def steps
  @steps
end

Instance Method Details

#run ⇒ Object

Run the transformation



26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
 
# File 'lib/rodimus/transformation.rb', line 26

def run
  notify(self, :before_run)
  @drb_server = DRb.start_service(nil, shared_data) unless using_threads?
  ids.clear
  prepare

  steps.each do |step|
    ids << in_parallel do
      step.shared_data = step_shared_data
      step.run
    end
    step.close_descriptors unless using_threads?
  end
ensure
  cleanup
  notify(self, :after_run)
end

#to_s ⇒ Object 



44
45
46
 
# File 'lib/rodimus/transformation.rb', line 44

def to_s
  "#{self.class} with #{steps.length} steps"
end