Improve reaping logic.

lib/async/node.rb

@@ -36,8 +36,10 @@ def initialize(parent = nil)
 
 		# Attach this node to an existing parent.
 		def parent=(parent)
+			return if @parent.equal?(parent)
+
 			if @parent
-				@parent.children.delete(self)
+				@parent.reap(self)
 				@parent = nil
 			end
 
@@ -47,16 +49,16 @@ def parent=(parent)
 			end
 		end
 
-		# Fold this node into it's parent, merging all children up.
 		def consume
-			raise RuntimeError.new("Cannot consume top level node") unless @parent
-
-			@children.each do |child|
-				# TODO: We could probably make this a bit more efficient.
-				child.parent = @parent
+			if @parent && @children.empty?
+				@parent.reap(self)
+				@parent.consume
+				@parent = nil
 			end
-
-			self.parent = nil
+		end
+
+		def reap(child)
+			@children.delete(child)
 		end
 	end
 end

lib/async/task.rb

@@ -54,7 +54,7 @@ def initialize(ios, reactor)
 				rescue Interrupt
 					# Async.logger.debug("Task #{self} interrupted: #{$!}")
 				ensure
-					consume
+					close
 				end
 			end
 		end
@@ -111,10 +111,11 @@ def self.current?
 			Thread.current[:async_task]
 		end
 
-		def consume
+		def close
 			@ios.each_value(&:close)
+			@ios = []
 
-			super
+			consume
 		end 
 
 		def inspect

spec/async/node_spec.rb

@@ -35,22 +35,25 @@
 			expect(child.parent).to be_nil
 			expect(subject.children).to be_empty
 		end
+
+		it "can consume bottom to top" do
+			child.consume
+
+			expect(child.parent).to be_nil
+			expect(subject.children).to be_empty
+		end
 	end
 
 	describe '#consume' do
 		let(:middle) {Async::Node.new(subject)}
 		let(:bottom) {Async::Node.new(middle)}
 
-		it "should merge child into parent" do
+		it "can't consume middle node" do
 			expect(bottom.parent).to be middle
 
 			middle.consume
 
-			expect(middle.parent).to be_nil
-			expect(middle.children).to be_empty
-
-			expect(bottom.parent).to be subject
-			expect(subject.children).to include(bottom)
+			expect(bottom.parent).to be middle
 		end
 	end
 end