Ensure Call queue is closed when dispatch exits.

Author: samuel-williams-shopify

lib/async/container/supervisor/connection.rb

@@ -147,10 +147,13 @@ def forward(target_connection, operation)
 					def self.dispatch(connection, target, id, message)
 						Async do
 							call = self.new(connection, id, message)
+							# Track the call in the connection's calls hash:
 							connection.calls[id] = call
 
+							# Dispatch the call to the target (synchronously):
 							target.dispatch(call)
 
+							# Stream responses back to the connection (asynchronously):
 							while response = call.pop
 								connection.write(id: id, **response)
 							end
@@ -160,6 +163,9 @@ def self.dispatch(connection, target, id, message)
 
 							# If the queue is closed, we don't need to send a finished message:
 							unless call.closed?
+								# Ensure the call is closed, to prevent messages being buffered:
+								call.close
+								
 								# If the above write failed, this is likely to fail too, and we can safely ignore it.
 								connection.write(id: id, finished: true) rescue nil
 							end
@@ -276,16 +282,19 @@ def call(...)
 				#
 				# @parameter target [Dispatchable] The target to dispatch calls to.
 				def run(target)
+					# Process incoming messages from the connection:
 					self.each do |message|
+						# If the message has an ID, it is a response to a call:
 						if id = message.delete(:id)
+							# Find the call in the connection's calls hash:
 							if call = @calls[id]
-								# Response to a call:
+								# Enqueue the response for the call:
 								call.push(**message)
 							elsif message.key?(:do)
-								# Incoming call:
+								# Otherwise, if we couldn't find an existing call, it must be a new call:
 								Call.dispatch(self, target, id, message)
 							else
-								# Likely a response to a timed-out call, ignore it:
+								# Finally, if none of the above, it is likely a response to a timed-out call, so ignore it:
 								Console.debug(self, "Ignoring message:", message)
 							end
 						else

releases.md

@@ -1,5 +1,9 @@
 # Releases
 
+## Unreleased
+
+  - Close `Call` queue if asynchronous call fails during dispatch - further messages will fail with `ClosedQueueError`.
+
 ## v0.9.0
 
   - Better handling of write failures in `Connection::Call.dispatch`, ensuring we don't leak calls.

test/async/container/connection.rb

@@ -40,6 +40,30 @@ def dispatch(call)
 			expect(connection.calls).to be(:empty?)
 		end
 
+		it "closes the queue when the connection fails" do
+			stream.write(JSON.dump({id: 1, do: :test}) << "\n")
+			stream.rewind
+			
+			expect(stream).to receive(:write).and_raise(IOError, "Test error")
+			
+			task = nil
+			
+			target = TestTarget.new do |call|
+				task = Async do
+					while true
+						call.push(status: "working")
+						sleep(1) # Loop forever (until the queue is closed).
+					end
+				end
+			end
+			
+			connection.run(target)
+			
+			expect(connection.calls).to be(:empty?)
+			expect(task).to be(:failed?)
+			expect{task.wait}.to raise_exception(ClosedQueueError)
+		end
+		
 		it "handles failed writes when making a call" do
 			expect(stream).to receive(:write).and_raise(IOError, "Test error")