100% test coverage.

Author: samuel-williams-shopify

lib/async/bus/protocol/transaction.rb

@@ -90,22 +90,27 @@ def invoke(name, arguments, options, &block)
 
 					self.write(Invoke.new(@id, name, arguments, options, block_given?))
 
-					while response = self.read
-						case response
-						when Return
-							return response.result
-						when Yield
-							begin
-								result = yield(*response.result)
-								self.write(Next.new(@id, result))
-							rescue => error
-								self.write(Error.new(@id, error))
-							end
-						when Error
-							raise(response.result)
+				while response = self.read
+					case response
+					when Return
+						return response.result
+					when Yield
+						begin
+							result = yield(*response.result)
+							self.write(Next.new(@id, result))
+						rescue => error
+							self.write(Error.new(@id, error))
 						end
+					when Error
+						raise(response.result)
+					when Throw
+						# Re-throw the tag and value that was thrown on the server side
+						# Throw.result contains [tag, value] array
+						tag, value = response.result
+						throw(tag, value)
 					end
 				end
+				end
 
 				# Accept a remote procedure invocation.
 				# @parameter object [Object] The object to invoke the method on.
@@ -134,7 +139,9 @@ def accept(object, arguments, options, block_given)
 
 					self.write(Return.new(@id, result))
 				rescue UncaughtThrowError => error
-					self.write(Throw.new(@id, error.tag))
+					# UncaughtThrowError has both tag and value attributes
+					# Store both in the Throw message: result is tag, we'll add value handling
+					self.write(Throw.new(@id, [error.tag, error.value]))
 				rescue => error
 					self.write(Error.new(@id, error))
 				end

test/async/bus/protocol/connection.rb

@@ -213,5 +213,160 @@ def get_temporary_object
 			end
 		end
 	end
+	
+	with "#[]=" do
+		it "can bind objects explicitly" do
+			server_connection = nil
+			
+			start_server do |connection|
+				server_connection = connection
+				object = Object.new
+				connection[:test] = object
+			end
+			
+			client.connect do |connection|
+				# Connect to trigger server's accept block
+				expect(server_connection.objects[:test]).to be_a(Async::Bus::Protocol::Connection::Explicit)
+				expect(server_connection.objects[:test].object).to be_a(Object)
+				expect(server_connection.objects[:test]).not.to be(:temporary?)
+			end
+		end
+	end
+	
+	with "#run" do
+		it "handles unexpected messages" do
+			error_logged = false
+			error_args = nil
+			
+			# Intercept Console.error to verify it's called
+			original_error = Console.method(:error)
+			Console.define_singleton_method(:error) do |*args|
+				error_logged = true
+				error_args = args
+				original_error.call(*args)
+			end
+			
+			begin
+				# Create a connection directly with a mock peer
+				peer = StringIO.new
+				connection = Async::Bus::Protocol::Connection.server(peer)
+				connection.bind(:test, Object.new)
+				
+				# Create an unexpected message object
+				unexpected_message = Object.new
+				
+				# Mock the unpacker to yield the unexpected message
+				original_unpacker = connection.instance_variable_get(:@unpacker)
+				mock_enumerator = Enumerator.new do |yielder|
+					yielder.yield(unexpected_message)
+					# Raise to stop the loop
+					raise IOError, "End of stream"
+				end
+				
+				connection.instance_variable_set(:@unpacker, mock_enumerator)
+				
+				# Run the connection in a task
+				task = Async do
+					begin
+						connection.run
+					rescue IOError
+						# Expected when enumerator raises
+					end
+				end
+				
+				# Wait for it to process
+				sleep(0.05)
+				
+				# Stop the task
+				task.stop
+				
+				expect(error_logged).to be_truthy
+				expect(error_args).not.to be_nil
+			ensure
+				# Restore original Console.error
+				Console.define_singleton_method(:error, original_error)
+			end
+		end
+		
+		it "closes pending transactions in ensure block when run loop exits" do
+			transactions_closed_count = 0
+			
+			# Create a connection directly to test the ensure block
+			peer = StringIO.new
+			connection = Async::Bus::Protocol::Connection.server(peer)
+			
+			# Create some transactions manually
+			transaction1 = connection.transaction!
+			transaction2 = connection.transaction!
+			
+			# Mock transaction.close to track calls
+			[transaction1, transaction2].each do |transaction|
+				original_close = transaction.method(:close)
+				transaction.define_singleton_method(:close) do
+					transactions_closed_count += 1
+					original_close.call
+				end
+			end
+			
+			# Verify transactions exist
+			expect(connection.transactions.size).to be == 2
+			
+			# Mock the unpacker to immediately raise (simulating connection close)
+			# This will trigger the ensure block
+			mock_enumerator = Enumerator.new do |yielder|
+				raise IOError, "Connection closed"
+			end
+			
+			connection.instance_variable_set(:@unpacker, mock_enumerator)
+			
+			# Run the connection - it will immediately hit the error and run ensure block
+			task = Async do
+				begin
+					connection.run
+				rescue IOError
+					# Expected
+				end
+			end
+			
+			# Wait for it to process
+			sleep(0.05)
+			
+			# Stop the task
+			task.stop
+			
+			# Verify transactions were closed in ensure block
+			expect(transactions_closed_count).to be == 2
+			expect(connection.transactions).to be(:empty?)
+		end
+		
+		it "closes pending transactions on connection close" do
+			start_server do |connection|
+				service = Object.new
+				def service.slow_method
+					sleep(0.1)
+					:result
+				end
+				
+				connection.bind(:service, service)
+			end
+			
+			client.connect do |connection|
+				# Start a transaction
+				transaction = connection.transaction!
+				
+				# Start an async call that will take time
+				Async do
+					connection[:service].slow_method
+				end
+				
+				# Close the connection immediately
+				connection.close
+				
+				# Verify transaction was closed
+				expect(transaction.connection).to be_nil
+				expect(transaction.received).to be(:closed?)
+			end
+		end
+	end
 end
 

test/async/bus/protocol/proxy.rb

@@ -0,0 +1,57 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2025, by Samuel Williams.
+
+require "async/bus/a_server"
+
+describe Async::Bus::Protocol::Proxy do
+	include Async::Bus::AServer
+	
+	with "operators" do
+		it "can use logical negation operator" do
+			start_server do |connection|
+				service = Object.new
+				def service.!
+					false
+				end
+				
+				connection.bind(:service, service)
+			end
+			
+			client.connect do |connection|
+				proxy = connection[:service]
+				result = !proxy
+				
+				expect(result).to be == false
+			end
+		end
+	end
+	
+	with "#respond_to_missing?" do
+		it "can check if remote object responds to method" do
+			start_server do |connection|
+				service = Object.new
+				def service.test_method
+					:result
+				end
+				
+				connection.bind(:service, service)
+			end
+			
+			client.connect do |connection|
+				proxy = connection[:service]
+				
+				# Call respond_to_missing? using instance_eval since Proxy inherits from BasicObject
+				# which doesn't have send/__send__/method
+				# (respond_to? is defined on Proxy, so Ruby won't call respond_to_missing? automatically)
+				result1 = proxy.instance_eval { respond_to_missing?(:test_method, false) }
+				result2 = proxy.instance_eval { respond_to_missing?(:nonexistent_method, false) }
+				
+				expect(result1).to be == true
+				expect(result2).to be == false
+			end
+		end
+	end
+end
+

test/async/bus/protocol/transaction.rb

@@ -174,6 +174,68 @@ def service.error_method
 				expect(write_count).to be > 0
 			end
 		end
+		
+		it "handles Close message in yield block to break iteration" do
+			start_server do |connection|
+				service = Object.new
+				def service.yielding_method
+					yield 1
+					yield 2
+					yield 3
+					:done
+				end
+				
+				connection.bind(:service, service)
+			end
+			
+			client.connect do |connection|
+				results = []
+				
+				# Create a transaction and manually handle yields to send Close
+				transaction = connection.transaction!
+				transaction.invoke(:service, [:yielding_method], {}) do |*yield_args|
+					value = yield_args.first
+					results << value
+					
+					# After first yield, send Close to break the loop
+					if results.size == 1
+						connection.write(Async::Bus::Protocol::Close.new(transaction.id, nil))
+					end
+					
+					:ack
+				end
+				
+				# Close should break the loop, so we should only get one value
+				expect(results.size).to be == 1
+				expect(results.first).to be == 1
+			end
+		end
+		
+		it "handles Throw message from server" do
+			start_server do |connection|
+				service = Object.new
+				def service.throw_method
+					throw :some_tag
+				end
+				
+				connection.bind(:service, service)
+			end
+			
+			client.connect do |connection|
+				# The server should catch the UncaughtThrowError and send a Throw message
+				# The client should handle it by re-throwing
+				# Note: UncaughtThrowError only preserves the tag, not the value
+				thrown = false
+				result = catch(:some_tag) do
+					connection[:service].throw_method
+					thrown = true
+				end
+				
+				# catch returns nil when throw happens without a value
+				expect(thrown).to be == false
+				expect(result).to be_nil
+			end
+		end
 	end
 end