Support passing Async::Semaphore as the parent task for Async::Job::Processor::Redis::Server

lib/async/job/processor/redis/server.rb

@@ -53,11 +53,13 @@ def start!
 
 						@task = true
 
-						@parent.async(transient: true, annotation: self.class.name) do |task|
+						Async do |task|
 							@task = task
 
 							while true
-								self.dequeue(task)
+								@parent.async(transient: true, annotation: self.class.name) do
+									self.dequeue
+								end
 							end
 						ensure
 							@task = nil
@@ -120,19 +122,17 @@ def call(job)
 					# If the job fails for any reason, it will be retried.
 					#
 					# If you do not desire this behavior, you should catch exceptions in the delegate.
-					def dequeue(parent)
+					def dequeue
 						_id = @processing_list.fetch
 
-						parent.async do
-							id = _id; _id = nil
-
-							job = @coder.load(@job_store.get(id))
-							@delegate.call(job)
-							@processing_list.complete(id)
-						rescue => error
-							Console.error(self, "Job failed with error!", id: id, exception: error)
-							@processing_list.retry(id)
-						end
+						id = _id; _id = nil
+
+						job = @coder.load(@job_store.get(id))
+						@delegate.call(job)
+						@processing_list.complete(id)
+					rescue => error
+						Console.error(self, "Job failed with error!", id: id, exception: error)
+						@processing_list.retry(id)
 					ensure
 						@processing_list.retry(_id) if _id
 					end

test/async/job/processor/server.rb

@@ -5,6 +5,7 @@
 
 require "async"
 require "async/redis"
+require "async/semaphore"
 
 require "sus/fixtures/async/reactor_context"
 require "sus/fixtures/console"
@@ -80,8 +81,70 @@
 			expect(server.status_string).to be == "R=0 D=0 P=0/0"
 
 			server.call(job)
+			sleep 0.1 # Allow some time for the job to be processed.
 
 			expect(server.status_string).to be == "R=0 D=0 P=0/1"
 		end
 	end
+
+	with "concurrency limit" do
+		# Delegate that sleeps for 5 seconds to simulate slow job processing
+		let(:slow_delegate) do
+			Class.new do
+				def start
+				end
+
+				def stop
+				end
+
+				def call(job)
+					sleep 5
+				end
+			end.new
+		end
+
+		with "Async::Idler" do
+			let(:idler_server) {subject.new(slow_delegate, prefix:, resolution: 1)}
+
+			it "can process all jobs concurrently" do
+				idler_server.start
+
+				# Enqueue 10 jobs
+				10.times do |i|
+					idler_server.call({"data" => "job #{i}"})
+				end
+
+				# Give time for all jobs to be picked up
+				sleep 0.5
+
+				# With Async::Idler (unlimited concurrency), all 10 jobs should be in processing status
+				status = idler_server.status_string
+				expect(status).to be =~ /P=(10|[5-9])\//
+
+				idler_server.stop
+			end
+		end
+
+		with "Async::Semaphore" do
+			let(:semaphore_server) {subject.new(slow_delegate, prefix:, resolution: 1, parent: Async::Semaphore.new(2))}
+
+			it "can limit concurrent job processing to 2" do
+				semaphore_server.start
+
+				# Enqueue 10 jobs
+				10.times do |i|
+					semaphore_server.call({"data" => "job #{i}"})
+				end
+
+				# Give time for jobs to be picked up
+				sleep 0.5
+
+				# Only 2 jobs should be in processing status
+				status = semaphore_server.status_string
+				expect(status).to be =~ /P=2\//
+
+				semaphore_server.stop
+			end
+		end
+	end
 end