lib/datapipes.rb in datapipes-0.0.2 vs lib/datapipes.rb in datapipes-0.0.3

@@ -22,11 +22,11 @@
 
     consumer = run_comsumer
     runners.each(&:join)
 
     notify_resource_ending
-    consumer.join if consumer.status == "run"
+    graceful_down(consumer)
   end
 
   private
 
   def run_comsumer
@@ -35,16 +35,24 @@
         break if resource_ended? && @pipe.empty?
 
         data = @tube.run_all(@pipe.pull)
         @sink.run_all(data)
       end
+      Thread.current.kill
     end
   end
 
   def notify_resource_ending
     @flag.enq true
+    Thread.pass
   end
 
   def resource_ended?
     !@flag.empty?
+  end
+
+  def graceful_down(consumer)
+    sleep 0.1
+    consumer.kill if consumer.status == 'sleep'
+    consumer.join
   end
 end