Allow assigning to semaphore limit. (#215)

Author: ioquatix

lib/async/semaphore.rb

@@ -28,6 +28,25 @@ def initialize(limit = 1, parent: nil)
 		# The tasks waiting on this semaphore.
 		attr :waiting
 
+		# Allow setting the limit. This is useful for cases where the semaphore is used to limit the number of concurrent tasks, but the number of tasks is not known in advance or needs to be modified.
+		#
+		# On increasing the limit, some tasks may be immediately resumed. On decreasing the limit, some tasks may execute until the count is < than the limit. 
+		#
+		# @parameter limit [Integer] The new limit.
+		def limit= limit
+			difference = limit - @limit
+			@limit = limit
+			
+			# We can't suspend 
+			if difference > 0
+				difference.times do
+					break unless node = @waiting.first
+					
+					node.resume
+				end
+			end
+		end
+		
 		# Is the semaphore currently acquired?
 		def empty?
 			@count.zero?

test/async/semaphore.rb

@@ -151,5 +151,49 @@
 		end
 	end
 
+	with '#limit' do
+		it "has a default limit of 1" do
+			expect(semaphore).to have_attributes(limit: be == 1)
+			expect(semaphore).not.to be(:blocking?)
+		end
+	end
+	
+	with '#limit=' do
+		it "releases tasks when limit is increased" do
+			semaphore.acquire
+			expect(semaphore).to have_attributes(count: be == 1)
+			expect(semaphore).to be(:blocking?)
+			
+			task = Async do
+				semaphore.acquire
+			end
+			
+			semaphore.limit = 2
+			task.wait
+			
+			expect(semaphore.count).to be == 2
+		end
+		
+		it "blocks tasks when limit is decreased" do
+			semaphore.limit = 2
+			semaphore.acquire
+			semaphore.acquire
+			
+			expect(semaphore).to have_attributes(count: be == 2)
+			expect(semaphore).to be(:blocking?)
+			
+			task = Async do
+				semaphore.acquire
+			end
+			
+			semaphore.limit = 1
+			semaphore.release
+			semaphore.release
+			task.wait
+			
+			expect(semaphore.count).to be == 1
+		end
+	end
+	
 	it_behaves_like ChainableAsync
 end