transformation.rb in rodimus-0.1.1

- old
+ new

@@ -1,25 +1,34 @@
+require 'drb'
+
 module Rodimus
 
   class Transformation
-    attr_reader :steps
+    attr_reader :drb_server, :steps
 
+    # User-data accessible across all running steps.
+    attr_reader :shared_data
+
     def initialize
       @steps = []
+      @shared_data = {} # TODO: This needs to be thread safe
     end
 
     def run
+      @drb_server = DRb.start_service(nil, shared_data)
       prepare
 
       steps.each do |step|
         fork do
+          DRb.start_service # the parent DRb thread dies across the fork
+          step.shared_data = DRbObject.new_with_uri(drb_server.uri)
           step.run
         end
-        step.incoming && step.incoming.close
-        step.outgoing && step.outgoing.close
+        step.close_descriptors
       end
-
+    ensure
       Process.waitall
+      drb_server.stop_service
     end
 
     def to_s
       "#{self.class} with #{steps.length} steps"
     end