› Async › Source › Async › LimitedQueue

class LimitedQueue

A queue which limits the number of items that can be enqueued.

Signature
public

Since Async v1.

Definitions

def self.new(...)

  • private

Signature
private

This exists purely for emitting a warning.

Implementation

 
def self.new(...)
	warn("`require 'async/limited_queue'` to use `Async::LimitedQueue`.", uplevel: 1, category: :deprecated) if $VERBOSE
	
	super
end

def initialize(limit = 1, full: Notification.new, **options)

Create a new limited queue.

Signature
parameter　limit　Integer

The maximum number of items that can be enqueued.

parameter　full　Notification

The notification to use for signaling when the queue is full.

Implementation

 
def initialize(limit = 1, full: Notification.new, **options)
	super(**options)
	
	@limit = limit
	@full = full
end

attr :limit

Signature
attribute　Integer

The maximum number of items that can be enqueued.

def close

Close the queue, causing all waiting tasks to return nil. Any subsequent calls to Async::LimitedQueue#enqueue will raise an exception. Also signals all tasks waiting for the queue to be full.

Implementation

 
def close
	super
	
	while @full.waiting?
		@full.signal(nil)
	end
end

def limited?

Signature
returns　Boolean

Whether trying to enqueue an item would block.

Implementation

 
def limited?
	!@closed && @items.size >= @limit
end

def push(item)

Add an item to the queue.

If the queue is full, this method will block until there is space available.

Signature
parameter　item　Object

The item to add to the queue.

Implementation

 
def push(item)
	while limited?
		@full.wait
	end
	
	super
end

def enqueue(*items)

Add multiple items to the queue.

If the queue is full, this method will block until there is space available.

Signature
parameter　items　Array

The items to add to the queue.

Implementation

 
def enqueue(*items)
	while !items.empty?
		while limited?
			@full.wait
		end
		
		if @closed
			raise ClosedError, "Cannot enqueue items to a closed queue."
		end
		
		available = @limit - @items.size
		@items.concat(items.shift(available))
		
		@available.signal unless self.empty?
	end
end

def dequeue

Remove and return the next item from the queue.

If the queue is empty, this method will block until an item is available.

Signature
returns　Object

The next item in the queue.

Implementation

 
def dequeue
	item = super
	
	@full.signal
	
	return item
end

Discussion