100% test coverage.

Author: samuel-williams-shopify

gems.rb

@@ -24,6 +24,7 @@
 	gem "rubocop"
 
 	gem "sus-fixtures-async"
+	gem "sus-fixtures-console"
 
 	gem "bake-test"
 	gem "bake-test-external"

lib/async/job/processor/aggregate.rb

@@ -34,7 +34,7 @@ def flush(jobs)
 						@delegate.call(job)
 					end
 				rescue => error
-					Console::Event::Failure.for(error).emit(self, "Could not flush #{jobs.size} jobs.")
+					Console.error(self, "Could not flush #{jobs.size} jobs.", exception: error)
 				end
 
 				# Run the background processing loop that continuously processes job batches.

lib/async/job/processor/generic.rb

@@ -10,8 +10,8 @@ module Processor
 			# This processor acts as a simple wrapper around any object that responds to call, start, and stop methods.
 			class Generic
 				# Initialize a new generic processor.
-				# @parameter delegate [Object | nil] The delegate object that will handle job execution.
-				def initialize(delegate = nil)
+				# @parameter delegate [Object] The delegate object that will handle job execution.
+				def initialize(delegate)
 					@delegate = delegate
 				end
 

lib/async/job/processor/inline.rb

@@ -43,12 +43,12 @@ def call(job)
 
 				# Start the processor by delegating to the configured delegate.
 				def start
-					@delegate&.start
+					@delegate.start
 				end
 
 				# Stop the processor by delegating to the configured delegate.
 				def stop
-					@delegate&.stop
+					@delegate.stop
 				end
 			end
 		end

test/async/job/buffer.rb

@@ -0,0 +1,107 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2024, by Samuel Williams.
+
+require "async/job/buffer"
+require "sus/fixtures/async/reactor_context"
+
+describe Async::Job::Buffer do
+	include Sus::Fixtures::Async::ReactorContext
+	
+	let(:buffer) {subject.new}
+	
+	with "no delegate" do
+		it "can be initialized without a delegate" do
+			expect(buffer).to be_a(Async::Job::Buffer)
+		end
+		
+		it "starts with an empty queue" do
+			expect(buffer).to be(:empty?)
+		end
+		
+		it "can add jobs to the buffer" do
+			job = {"data" => "test job"}
+			buffer.call(job)
+			
+			expect(buffer).not.to be(:empty?)
+			expect(buffer.pop).to be == job
+		end
+		
+		it "can add multiple jobs" do
+			job1 = {"id" => 1}
+			job2 = {"id" => 2}
+			
+			buffer.call(job1)
+			buffer.call(job2)
+			
+			expect(buffer.pop).to be == job1
+			expect(buffer.pop).to be == job2
+			expect(buffer).to be(:empty?)
+		end
+				
+		it "can start without a delegate" do
+			expect{buffer.start}.not.to raise_exception
+		end
+		
+		it "can stop without a delegate" do
+			expect{buffer.stop}.not.to raise_exception
+		end
+	end
+	
+	with "delegate" do
+		let(:delegate) do
+			Class.new do
+				def initialize
+					@jobs = []
+					@started = false
+					@stopped = false
+				end
+				
+				attr_reader :jobs, :started, :stopped
+				
+				def call(job)
+					@jobs << job
+				end
+				
+				def start
+					@started = true
+				end
+				
+				def stop
+					@stopped = true
+				end
+			end.new
+		end
+		
+		let(:buffer) {subject.new(delegate)}
+		
+		it "can be initialized with a delegate" do
+			expect(buffer).to be_a(Async::Job::Buffer)
+		end
+		
+		it "delegates job calls to the delegate" do
+			job = {"data" => "test job"}
+			buffer.call(job)
+			
+			expect(delegate.jobs).to have_value(be == job)
+		end
+		
+		it "starts the delegate when start is called" do
+			buffer.start
+			expect(delegate.started).to be == true
+		end
+		
+		it "stops the delegate when stop is called" do
+			buffer.stop
+			expect(delegate.stopped).to be == true
+		end
+		
+		it "still maintains its own queue" do
+			job = {"data" => "test job"}
+			buffer.call(job)
+			
+			expect(buffer.pop).to be == job
+		end
+	end
+end 

test/async/job/builder.rb

@@ -25,4 +25,75 @@
 
 		expect(buffer.pop).to be == "My job"
 	end
+
+	it "can add dequeue middleware" do
+		buffer = Async::Job::Buffer.new
+		
+		# Create a pipeline with dequeue middleware:
+		pipeline = Async::Job::Builder.build(buffer) do
+			dequeue Async::Job::Processor::Inline
+		end
+		
+		# The dequeue middleware should be applied to the server side
+		expect(pipeline.server).to be_a(Async::Job::Processor::Inline)
+		expect(pipeline.client).to be_a(Async::Job::Processor::Inline)
+		expect(pipeline.delegate).to be == buffer
+	end
+
+	it "can set delegate using delegate method" do
+		custom_delegate = Object.new
+		
+		# Create a pipeline with custom delegate:
+		pipeline = Async::Job::Builder.build do
+			delegate custom_delegate
+		end
+		
+		expect(pipeline.delegate).to be == custom_delegate
+		expect(pipeline.client).to be == custom_delegate
+		expect(pipeline.server).to be == custom_delegate
+	end
+
+	it "can build with both enqueue and dequeue middleware" do
+		buffer = Async::Job::Buffer.new
+		
+		# Create a pipeline with both enqueue and dequeue middleware:
+		pipeline = Async::Job::Builder.build(buffer) do
+			enqueue Async::Job::Processor::Inline
+			dequeue Async::Job::Processor::Inline
+		end
+		
+		# Both client and server should be wrapped with Inline processors
+		expect(pipeline.client).to be_a(Async::Job::Processor::Inline)
+		expect(pipeline.server).to be_a(Async::Job::Processor::Inline)
+		expect(pipeline.delegate).to be == buffer
+	end
+
+	it "can build with custom delegate parameter" do
+		custom_delegate = Object.new
+		
+		# Create a pipeline with custom delegate passed to build:
+		pipeline = Async::Job::Builder.build(custom_delegate) do
+			enqueue Async::Job::Processor::Inline
+		end
+		
+		expect(pipeline.delegate).to be == custom_delegate
+		expect(pipeline.client).to be_a(Async::Job::Processor::Inline)
+		expect(pipeline.server).to be == custom_delegate
+	end
+
+	it "can chain middleware methods" do
+		buffer = Async::Job::Buffer.new
+		
+		# Test method chaining:
+		builder = Async::Job::Builder.new(buffer)
+		result = builder.enqueue(Async::Job::Processor::Inline)
+			.dequeue(Async::Job::Processor::Inline)
+			.delegate(buffer)
+		
+		expect(result).to be == builder
+		
+		# Build the pipeline:
+		pipeline = builder.build
+		expect(pipeline.delegate).to be == buffer
+	end
 end

test/async/job/processor/aggregate.rb

@@ -3,63 +3,84 @@
 # Released under the MIT License.
 # Copyright, 2024, by Samuel Williams.
 
-require "async"
-
+require "async/job/processor/aggregate"
 require "sus/fixtures/async/reactor_context"
+require "sus/fixtures/console/captured_logger"
 
-require "async/job/buffer"
-require "async/job/processor/aggregate"
+describe Async::Job::Processor::Aggregate do
+	include Sus::Fixtures::Async::ReactorContext
+	include_context Sus::Fixtures::Console::CapturedLogger
 
-class SlowQueue
-	def initialize(delegate = nil)
-		@condition = Async::Condition.new
-		@delegate = delegate
+	let(:delegate) do
+		Class.new do
+			attr_reader :called_jobs, :started, :stopped
+			def initialize
+				@called_jobs = []
+				@started = false
+				@stopped = false
+			end
+			def call(job)
+				@called_jobs << job
+			end
+			def start
+				@started = true
+				"started"
+			end
+			def stop
+				@stopped = true
+				"stopped"
+			end
+		end.new
 	end
-	
-	attr :condition
-	
-	def call(job)
-		@condition.wait
-		@delegate&.call(job)
+
+	let(:processor) {subject.new(delegate)}
+
+	it "processes jobs in batches" do
+		processor.call(:job1)
+		processor.call(:job2)
+		
+		# Give the processor a chance to process jobs
+		sleep(0.1)
+		
+		expect(delegate.called_jobs).to be(:include?, :job1)
+		expect(delegate.called_jobs).to be(:include?, :job2)
 	end
-	
-	def start
-		@delegate&.start
+
+	it "handles errors in flush" do
+		error_delegate = Class.new do
+			def call(job)
+				raise "Test error"
+			end
+		end.new
+		
+		error_processor = subject.new(error_delegate)
+		error_processor.call(:job1)
+		# Give the processor a chance to process jobs and handle error
+		sleep(0.1)
+		
+		# Assert that the error was logged
+		expect_console.to have_logged(
+			severity: be == :error,
+			message: be(:include?, "Could not flush")
+		)
 	end
-	
-	def stop
-		@delegate&.stop
+
+	it "delegates start to super and start!" do
+		# This will call super (Generic#start) and then start!
+		# We can't easily check super, but we can check that start! returns true
+		result = processor.send(:start!)
+		expect(result).to be == true
 	end
-end
 
-describe Async::Job::Processor::Aggregate do
-	include Sus::Fixtures::Async::ReactorContext
-	
-	let(:buffer) {Async::Job::Buffer.new}
-	let(:server) {subject.new(buffer)}
-	
-	let(:job) {{"data" => "test job"}}
-	
-	it "can schedule a job" do
-		server.call(job)
-		
-		expect(buffer.pop).to be == job
+	it "delegates start to delegate" do
+		result = processor.start
+		expect(delegate.started).to be == true
+		expect(result).to be == true
 	end
-	
-	with "slow queue" do
-		let(:queue) {SlowQueue.new(buffer)}
-		let(:server) {subject.new(queue)}
-		
-		it "flushes jobs on shutdown" do
-			server.call(job)
-			server.stop
-			
-			expect(buffer).to be(:empty?)
-			
-			# Allow job processing to continue:
-			queue.condition.signal
-			
-			expect(buffer.pop).to be == job
-		end
+
+	it "delegates stop to delegate" do
+		result = processor.stop
+		expect(delegate.stopped).to be == true
+		expect(result).to be == "stopped"
 	end
 end