WIP

Author: ioquatix

fixtures/async/a_queue.rb

@@ -20,6 +20,18 @@ module Async
 				expect(queue.size).to be == 1
 				expect(queue.dequeue).to be == :item
 			end
+			
+			it "can't add an item to a closed queue" do
+				queue.push(:item)
+				expect(queue).to have_attributes(size: be == 1)
+				expect(queue.dequeue).to be == :item
+				
+				queue.close
+				
+				expect do
+					queue.push(:item)
+				end.to raise_exception(Async::Queue::ClosedError)
+			end
 		end
 
 		with "#pop" do

lib/async/queue.rb

@@ -15,6 +15,11 @@ module Async
 	#
 	# @public Since *Async v1*.
 	class Queue
+		# An error raised when trying to enqueue items to a closed queue.
+		# @public Since *Async v2.24*.
+		class ClosedError < RuntimeError
+		end
+		
 		# Create a new queue.
 		#
 		# @parameter parent [Interface(:async) | Nil] The parent task to use for async operations.
@@ -26,6 +31,7 @@ def initialize(parent: nil, available: Notification.new)
 			@available = available
 		end
 
+		# Close the queue, causing all waiting tasks to return `nil`. Any subsequent calls to {enqueue} will raise an exception.
 		def close
 			@closed = true
 
@@ -61,6 +67,10 @@ def <<(item)
 
 		# Add multiple items to the queue.
 		def enqueue(*items)
+			if @closed
+				raise ClosedError, "Cannot enqueue items to a closed queue."
+			end
+			
 			@items.concat(items)
 
 			@available.signal unless self.empty?
@@ -133,6 +143,14 @@ def initialize(limit = 1, full: Notification.new, **options)
 		# @attribute [Integer] The maximum number of items that can be enqueued.
 		attr :limit
 
+		def close
+			super
+			
+			while @full.waiting?
+				@full.signal(nil)
+			end
+		end
+		
 		# @returns [Boolean] Whether trying to enqueue an item would block.
 		def limited?
 			@items.size >= @limit
@@ -162,6 +180,10 @@ def enqueue(*items)
 					@full.wait
 				end
 
+				if @closed
+					raise ClosedError, "Cannot enqueue items to a closed queue."
+				end
+				
 				available = @limit - @items.size
 				@items.concat(items.shift(available))