Reopen controller event queue for each run

Assisted-By: devx/3236e566-7538-432e-a30a-2bdf37265ed4

Author: samuel-williams-shopify

lib/async/container/controller.rb

@@ -64,6 +64,7 @@ def initialize(notify: Notify.open!, container_class: Container, graceful_stop:
 				@container = nil
 				@events = ::Thread::Queue.new
 				@signals = Async::Signals::Handlers.new
+				@running = false
 
 				# Serializes lifecycle transitions such as start, restart and reload. `Container#stop` (which can also take time) is performed outside this guard, so that live container events are not blocked by the stop operation (e.g. restarting).
 				@guard = ::Thread::Mutex.new
@@ -117,7 +118,7 @@ def trap(signal, &block)
 					event = SignalEvent.new(signal, block).freeze
 
 					@signals.trap(signal) do
-						@events << event
+						enqueue_event(event)
 					end
 				else
 					@signals.ignore(signal)
@@ -272,12 +273,44 @@ def reload
 				end
 			end
 
-			private def wait_for_container
+			private def enqueue_event(event)
+				@events << event
+			rescue ::ClosedQueueError
+				# The controller run loop has already stopped.
+			end
+			
+			private def open_event_queue
+				@guard.synchronize do
+					if @running
+						raise RuntimeError, "Controller is already running."
+					end
+					
+					@running = true
+					@events = ::Thread::Queue.new
+				end
+			end
+			
+			private def close_event_queue(events)
+				events.close
+			end
+			
+			private def finish_event_queue(events)
+				events.close
+				
+				@guard.synchronize do
+					if @events.equal?(events)
+						@running = false
+						@events = ::Thread::Queue.new
+					end
+				end
+			end
+			
+			private def wait_for_container(events)
 				while true
 					container = @guard.synchronize{@container}
 
 					if container.nil?
-						@events.close
+						close_event_queue(events)
 						return
 					end
 
@@ -287,7 +320,7 @@ def reload
 						# If this is still the active container, it completed naturally. Clear it and close the event queue so the controller run loop can finish. If it was replaced by a restart, keep waiting for the new active container.
 						if @container.equal?(container)
 							@container = nil
-							@events.close
+							close_event_queue(events)
 							return
 						end
 					end
@@ -298,14 +331,15 @@ def reload
 			# @parameter signals [#install] The signal backend to use while running the controller.
 			def run(signals: Async::Signals.default)
 				@notify&.status!("Initializing controller...")
+				events = open_event_queue
 
 				signals.install(@signals) do
 					Sync do |task|
 						self.start
 
-						waiter = task.async{wait_for_container}
+						task.async{wait_for_container(events)}
 
-						while event = @events.pop
+						while event = events.pop
 							event.call
 						end
 					rescue Async::Cancel
@@ -318,6 +352,8 @@ def run(signals: Async::Signals.default)
 				end
 			rescue Interrupt
 				self.stop(false)
+			ensure
+				finish_event_queue(events) if events
 			end
 		end
 	end

test/async/container/controller.rb

@@ -187,6 +187,35 @@ def controller.setup(container)
 		end
 	end
 
+	with "#run" do
+		it "can run the same controller more than once" do
+			input, output = IO.pipe
+			
+			controller.instance_variable_set(:@output, output)
+			
+			def controller.setup(container)
+				container.spawn do |instance|
+					instance.ready!
+					
+					sleep(0.01)
+					
+					@output.puts("done")
+					@output.flush
+				end
+			end
+			
+			2.times do
+				controller.run(signals: Async::Signals::Ignore)
+				
+				expect(IO.select([input], nil, nil, 1)).not.to be_nil
+				expect(input.gets).to be == "done\n"
+			end
+		ensure
+			input&.close
+			output&.close
+		end
+	end
+	
 	with "graceful controller" do
 		include_context Async::Container::AController, "graceful"