lib/datapipes/source.rb in datapipes-0.1.2 vs lib/datapipes/source.rb in datapipes-0.1.3

@@ -12,15 +12,20 @@
 
     attr_accessor :pipe
 
     def run_all
       @accumulated ||= [self]
+      set_pipe
       @accumulated.map {|s| Thread.new { s.run } }
     end
 
     private
 
     def produce(data)
-      pipe.recieve(data)
+      @pipe.recieve(data)
+    end
+
+    def set_pipe
+      @accumulated.each {|s| s.pipe = @pipe }
     end
   end
 end