lib/async/http/body.rb in async-http-0.16.0 vs lib/async/http/body.rb in async-http-0.17.0

@@ -20,35 +20,43 @@
 
 require 'async/queue'
 
 module Async
 	module HTTP
-		class Body < Async::Queue
+		class Body
 			def initialize
-				super
+				@queue = Async::Queue.new
 				
 				@finished = false
+				@stopped = false
 			end
 			
 			def finished?
 				@finished
 			end
 			
 			def each
+				return to_enum unless block_given?
+				
 				return if @finished
 				
-				while chunk = self.dequeue
+				while chunk = @queue.dequeue
 					yield chunk
 				end
+			rescue
+				# Stop the stream because the remote end is no longer reading from it. Any attempt to write to the stream will fail.
+				@stopped = $!
 				
+				raise
+			ensure
 				@finished = true
 			end
 			
 			def read
 				return if @finished
 				
-				unless chunk = self.dequeue
+				unless chunk = @queue.dequeue
 					@finished = true
 				end
 				
 				return chunk
 			end
@@ -61,18 +69,22 @@
 				end
 				
 				return buffer
 			end
 			
-			alias join read
-			
 			def write(chunk)
-				self.enqueue(chunk)
+				if @stopped
+					raise @stopped
+				end
+				
+				# TODO should this yield if the queue is full?
+				
+				@queue.enqueue(chunk)
 			end
 			
 			def finish
-				self.enqueue(nil)
+				@queue.enqueue(nil)
 			end
 		end
 		
 		class BufferedBody
 			def initialize(body)
@@ -214,11 +226,9 @@
 				
 				@remaining = 0
 				
 				return buffer
 			end
-			
-			alias join read
 			
 			def finish
 				read
 			end
 		end