Add a concurrency primitive for waiting for a specific number of tasks to complete.

Author: ioquatix

lib/async/waiter.rb

@@ -0,0 +1,33 @@
+
+module Async
+	# A composable synchronization primitive, which allows one task to wait for a number of other tasks to complete. It can be used in conjunction with {Semaphore} and/or {Barrier}.
+	class Waiter
+		def initialize(parent: nil, finished: Async::Condition.new)
+			@finished = finished
+			@done = []
+			
+			@parent = parent
+		end
+		
+		def async(parent: (@parent or Task.current), &block)
+			parent.async do |task|
+				yield(task)
+			ensure
+				@done << task
+				@finished.signal
+			end
+		end
+		
+		def wait_for(count)
+			while @done.size < count
+				@finished.wait
+			end
+			
+			return @done.shift(count)
+		end
+		
+		def wait(count, exception: false)
+			wait_for(count).map(&:result)
+		end
+	end
+end

test/async/waiter.rb

@@ -0,0 +1,23 @@
+
+require 'async/waiter'
+require 'sus/fixtures/async'
+
+describe "Async::Waiter" do
+	include Sus::Fixtures::Async::ReactorContext
+	
+	it "can wait for a subset of tasks" do
+		waiter = Async::Waiter.new
+		
+		3.times do
+			waiter.async do
+				sleep(rand * 0.01)
+			end
+		end
+		
+		done = waiter.wait(2)
+		expect(done.size).to be == 2
+
+		done = waiter.wait(1)
+		expect(done.size).to be == 1
+	end
+end