Reorganize implementation of Async::LimitedQueue and related tests.

Author: ioquatix

fixtures/async/a_queue.rb

@@ -0,0 +1,182 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2018-2024, by Samuel Williams.
+# Copyright, 2019, by Ryan Musgrave.
+# Copyright, 2020-2022, by Bruno Sutic.
+
+require "async"
+require "async/queue"
+require "sus/fixtures/async"
+require "async/semaphore"
+
+require "async/chainable_async"
+
+module Async
+	AQueue = Sus::Shared("a queue") do
+		let(:queue) {subject.new}
+		
+		with "#push" do
+			it "adds an item to the queue" do
+				queue.push(:item)
+				expect(queue.size).to be == 1
+				expect(queue.dequeue).to be == :item
+			end
+		end
+		
+		with "#pop" do
+			it "removes an item from the queue" do
+				queue.push(:item)
+				expect(queue.pop).to be == :item
+				expect(queue.size).to be == 0
+			end
+		end
+		
+		with "#each" do
+			it "can enumerate queue items" do
+				reactor.async do |task|
+					10.times do |item|
+						task.sleep(0.0001)
+						queue.enqueue(item)
+					end
+					
+					queue.enqueue(nil)
+				end
+				
+				items = []
+				queue.each do |item|
+					items << item
+				end
+				
+				expect(items).to be == 10.times.to_a
+			end
+		end
+		
+		it "should process items in order" do
+			reactor.async do |task|
+				10.times do |i|
+					task.sleep(0.001)
+					queue.enqueue(i)
+				end
+			end
+			
+			10.times do |j|
+				expect(queue.dequeue).to be == j
+			end
+		end
+		
+		it "can enqueue multiple items" do
+			items = Array.new(10) { rand(10) }
+
+			reactor.async do |task|
+				queue.enqueue(*items)
+			end
+
+			items.each do |item|
+				expect(queue.dequeue).to be == item
+			end
+		end
+		
+		it "can dequeue items asynchronously" do
+			reactor.async do |task|
+				queue << 1
+				queue << nil
+			end
+			
+			queue.async do |task, item|
+				expect(item).to be == 1
+			end
+		end
+		
+		with "#<<" do
+			it "adds an item to the queue" do
+				queue << :item
+				expect(queue.size).to be == 1
+				expect(queue.dequeue).to be == :item
+			end
+		end
+		
+		with "#size" do
+			it "returns queue size" do
+				expect(queue.size).to be == 0
+				queue.enqueue("Hello World")
+				expect(queue.size).to be == 1
+			end
+		end
+		
+		with "#signal" do
+			it "can signal with an item" do
+				queue.signal(:item)
+				expect(queue.dequeue).to be == :item
+			end
+		end
+		
+		with "#wait" do
+			it "can wait for an item" do
+				reactor.async do |task|
+					queue.enqueue(:item)
+				end
+				
+				expect(queue.wait).to be == :item
+			end
+		end
+		
+		with "an empty queue" do
+			it "is expected to be empty" do
+				expect(queue).to be(:empty?)
+			end
+		end
+		
+		with "task finishing queue" do
+			it "can signal task completion" do
+				3.times do
+					Async(finished: queue) do
+						:result
+					end
+				end
+				
+				3.times do
+					task = queue.dequeue
+					expect(task.wait).to be == :result
+				end
+			end
+		end
+		
+		with "semaphore" do
+			let(:capacity) {2}
+			let(:semaphore) {Async::Semaphore.new(capacity)}
+			let(:repeats) {capacity * 2}
+			
+			it "should process several items limited by a semaphore" do
+				count = 0
+				
+				Async do
+					repeats.times do
+						queue.enqueue :item
+					end
+					
+					queue.enqueue nil
+				end
+				
+				queue.async(parent: semaphore) do |task|
+					count += 1
+				end
+				
+				expect(count).to be == repeats
+			end
+		end
+		
+		it_behaves_like Async::ChainableAsync do
+			def before
+				chainable.enqueue(:item)
+				
+				# The limited queue may block.
+				Async do
+					chainable.enqueue(nil)
+				end
+				
+				super
+			end
+		end
+	end
+end

fixtures/async/chainable_async.rb

@@ -0,0 +1,19 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2019-2024, by Samuel Williams.
+
+module Async
+	ChainableAsync = Sus::Shared("chainable async") do
+		let(:parent) {Object.new}
+		let(:chainable) {subject.new(parent: parent)}
+		
+		it "should chain async to parent" do
+			expect(parent).to receive(:async).and_return(nil)
+			
+			chainable.async do
+				# Nothing.
+			end
+		end
+	end
+end

fixtures/chainable_async.rb

@@ -29,7 +29,8 @@
 	gem "metrics"
 
 	gem "sus-fixtures-async"
-	gem "sus-fixtures-console", "~> 0.3"
+	gem "sus-fixtures-console"
+	gem "sus-fixtures-time"
 
 	gem "bake-test"
 	gem "bake-test-external"

fixtures/timer_quantum.rb

@@ -0,0 +1,7 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2024, by Samuel Williams.
+
+# The implementation lives in `queue.rb` but later we may move it here for better autoload/inference.
+require_relative "queue"

gems.rb

@@ -8,8 +8,7 @@
 require "sus/fixtures/async"
 require "async/semaphore"
 
-require "chainable_async"
-require "timer_quantum"
+require "async/chainable_async"
 
 describe Async::Barrier do
 	include Sus::Fixtures::Async::ReactorContext
@@ -40,7 +39,7 @@
 
 			duration = Async::Clock.measure{barrier.wait}
 
-			expect(duration).to be_within(Q*repeats).of(delay)
+			expect(duration).to be_within(repeats * Sus::Fixtures::Time::QUANTUM).of(delay)
 			expect(finished).to be == repeats
 			expect(barrier).to be(:empty?)
 		end
@@ -186,5 +185,5 @@
 		end
 	end
 
-	it_behaves_like ChainableAsync
+	it_behaves_like Async::ChainableAsync
 end

lib/async/limited_queue.rb

@@ -4,8 +4,7 @@
 # Copyright, 2018-2024, by Samuel Williams.
 
 require "async/clock"
-
-require "timer_quantum"
+require "sus/fixtures/time/quantum"
 
 describe Async::Clock do
 	let(:clock) {subject.new}
@@ -15,7 +14,7 @@
 			sleep 0.01
 		end
 
-		expect(duration).to be_within(Q).of(0.01)
+		expect(duration).to be_within(Sus::Fixtures::Time::QUANTUM).of(0.01)
 	end
 
 	it "can get current offset" do
@@ -29,7 +28,7 @@
 			clock.stop!
 		end
 
-		expect(clock.total).to be_within(2 * Q).of(0.02)
+		expect(clock.total).to be_within(2 * Sus::Fixtures::Time::QUANTUM).of(0.02)
 	end
 
 	with "#total" do

test/async/barrier.rb

@@ -6,7 +6,7 @@
 require "async/idler"
 require "sus/fixtures/async"
 
-require "chainable_async"
+require "async/chainable_async"
 
 describe Async::Idler do
 	include Sus::Fixtures::Async::ReactorContext
@@ -33,5 +33,5 @@
 		expect(Fiber.scheduler.load).to be_within(0.1).of(0.5)
 	end
 
-	it_behaves_like ChainableAsync
+	it_behaves_like Async::ChainableAsync
 end