Multi-hop invoke.

Author: samuel-williams-shopify

guides/controllers/readme.md

@@ -112,10 +112,10 @@ When a controller method returns another controller, the client receives a proxy
 
 ```ruby
 # Server:
-server = ChatServerController.new
+chat = ChatServerController.new
 
 server.accept do |connection|
-	connection.bind(:chat, server)
+	connection.bind(:chat, chat)
 end
 
 # Client 1:
@@ -135,7 +135,7 @@ end
 
 ## Passing Controllers as Arguments
 
-Because controllers are passed by reference, you can pass them as arguments to enable bidirectional communication. This pattern is useful for event handlers, callbacks, or subscription systems:
+Because controllers are passed by reference, you can pass them as arguments to enable bidirectional communication. When a client passes a proxy as an argument, the server receives a proxy that points back to the client's controller. This enables the server to call methods on the client's controller. This pattern is useful for event handlers, callbacks, or subscription systems:
 
 ```ruby
 class ChatRoomController < Async::Bus::Controller
@@ -149,7 +149,7 @@ class ChatRoomController < Async::Bus::Controller
 		# subscriber is a proxy to the client's controller:
 		@subscribers << subscriber
 		# Send existing messages to the new subscriber:
-		@messages.each { |msg| subscriber.on_message(msg) }
+		@messages.each{|msg| subscriber.on_message(msg)}
 		true
 	end
 

guides/getting-started/readme.md

@@ -35,7 +35,7 @@ Async do
 
 	# Shared mutable state:
 	items = Array.new
-
+	
 	server.accept do |connection|
 		# Bind any object - it will be proxied to clients:
 		connection.bind(:items, items)

lib/async/bus/client.rb

@@ -64,13 +64,15 @@ def connect(parent: Task.current)
 			# This is useful for long-running clients that need to maintain a persistent connection.
 			#
 			# @parameter parent [Async::Task] The parent task to run under.
-			def run(parent: Task.current)
-				parent.async(annotation: "Bus Client", transient: true) do |task|
+			def run
+				Sync do |task|
 					loop do
 						connection = connect!
 
 						connected_task = task.async do
 							connected!(connection)
+							
+							yield(connection) if block_given?
 						end
 
 						connection.run

lib/async/bus/protocol/connection.rb

@@ -143,19 +143,14 @@ def []=(name, object)
 
 				# Generate a proxy for a remotely bound object.
 				#
-				# **This will not return objects bound locally, only proxies for remotely bound objects.**
+				# **This always returns a proxy, even if the object is bound locally.**
+				# The object bus is not shared between client and server, so `[]` always
+				# returns a proxy to the remote instance.
 				#
 				# @parameter name [String] The name of the bound object.
-				# @returns [Object | Proxy] The object or proxy instance for the bound object.
+				# @returns [Proxy] A proxy instance for the bound object.
 				def [](name)
-					unless proxy = @proxies[name]
-						proxy = Proxy.new(self, name)
-						@proxies[name] = proxy
-						
-						::ObjectSpace.define_finalizer(proxy, finalize(name))
-					end
-					
-					return proxy
+					return proxy_for(name)
 				end
 
 				# Explicitly bind an object to a name, such that it could be accessed remotely.
@@ -175,8 +170,8 @@ def bind(name, object)
 					# Bind the object into the local object store (explicitly bound, not temporary):
 					@objects[name] = Explicit.new(object)
 
-					# Return the proxy instance for the bound object:
-					return self[name]
+					# Always return a proxy for passing by reference, even for locally bound objects:
+					return proxy_for(name)
 				end
 
 				# Implicitly bind an object with a temporary name, such that it could be accessed remotely.
@@ -188,13 +183,13 @@ def bind(name, object)
 				# @parameter object [Object] The object to bind to a temporary name.
 				# @returns [Proxy] A proxy instance for the bound object.
 				def proxy(object)
-					name = "<#{object.class}@#{next_id.to_s(16)}>".freeze
+					name = object.__id__
 
 					# Bind the object into the local object store (temporary):
-					@objects[name] = Implicit.new(object)
+					@objects[name] ||= Implicit.new(object)
 
-					# This constructs the Proxy instance:
-					return self[name]
+					# Always return a proxy for passing by reference:
+					return proxy_for(name)
 				end
 
 				# Implicitly bind an object with a temporary name, such that it could be accessed remotely.
@@ -206,15 +201,50 @@ def proxy(object)
 				# @parameter object [Object] The object to bind to a temporary name.
 				# @returns [String] The name of the bound object.
 				def proxy_name(object)
-					name = "<#{object.class}@#{next_id.to_s(16)}>".freeze
+					name = object.__id__
 
 					# Bind the object into the local object store (temporary):
-					@objects[name] = Implicit.new(object)
+					@objects[name] ||= Implicit.new(object)
 
 					# Return the name:
 					return name
 				end
 
+				# Get an object or proxy for a bound object, handling reverse lookup.
+				#
+				# If the object is bound locally and the proxy is for this connection, returns the actual object.
+				# If the object is bound remotely, or the proxy is from a different connection, returns a proxy.
+				# This is used when deserializing proxies to handle round-trip scenarios and avoid name collisions.
+				#
+				# @parameter name [String] The name of the bound object.
+				# @parameter local [Boolean] Whether the proxy is for this connection (from serialization). Defaults to true.
+				# @returns [Object | Proxy] The object if bound locally and proxy is for this connection, or a proxy otherwise.
+				def proxy_object(name)
+					# If the proxy is for this connection and the object is bound locally, return the actual object:
+					if entry = @objects[name]
+						# This handles round-trip scenarios correctly.
+						return entry.object
+					end
+					
+					# Otherwise, create a proxy for the remote object:
+					return proxy_for(name)
+				end
+				
+				# Get or create a proxy for a named object.
+				#
+				# @parameter name [String] The name of the object.
+				# @returns [Proxy] A proxy instance for the named object.
+				private def proxy_for(name)
+					unless proxy = @proxies[name]
+						proxy = Proxy.new(self, name)
+						@proxies[name] = proxy
+						
+						::ObjectSpace.define_finalizer(proxy, finalize(name))
+					end
+					
+					return proxy
+				end
+				
 				private def finalize(name)
 					proc do
 						@finalized.push(name) rescue nil

lib/async/bus/protocol/proxy.rb

@@ -19,6 +19,12 @@ def initialize(connection, name)
 					@name = name
 				end
 
+				# Get the connection to the remote object.
+				# @returns [Connection] The connection to the remote object.
+				def __connection__
+					@connection
+				end
+				
 				# Get the name of the remote object.
 				# @returns [Symbol] The name of the remote object.
 				def __name__

lib/async/bus/protocol/wrapper.rb

@@ -26,54 +26,91 @@ def initialize(connection, reference_types: [Controller])
 					@connection = connection
 					@reference_types = reference_types
 
+					# Store the peer connection for forwarding proxies:
+					# When a proxy is forwarded (local=false), it should point back to the sender
+					# (the peer connection), not the receiver (this connection).
+					@peer_connection = nil
+					
 					# The order here matters.
 
 					self.register_type(0x00, Invoke, recursive: true,
-						packer: ->(invoke, packer){invoke.pack(packer)},
-						unpacker: ->(unpacker){Invoke.unpack(unpacker)},
-					)
+							packer: ->(invoke, packer){invoke.pack(packer)},
+							unpacker: ->(unpacker){Invoke.unpack(unpacker)},
+						)
 
 					[Return, Yield, Error, Next, Throw, Close].each_with_index do |klass, index|
 						self.register_type(0x01 + index, klass, recursive: true,
-							packer: ->(value, packer){value.pack(packer)},
-							unpacker: ->(unpacker){klass.unpack(unpacker)},
-						)
+								packer: ->(value, packer){value.pack(packer)},
+								unpacker: ->(unpacker){klass.unpack(unpacker)},
+							)
 					end
 
 					# Reverse serialize proxies back into proxies:
-					# When a Proxy is received, create a proxy pointing back
+					# When a Proxy is received, use proxy_object to handle reverse lookup
 					self.register_type(0x10, Proxy, recursive: true,
-						# Since name can be a Symbol or String, we need to use recursive packing to ensure the name is properly serialized.
-						packer: ->(proxy, packer){packer.write(proxy.__name__)},
-						unpacker: ->(unpacker){connection[unpacker.read]},
-					)
+							packer: self.method(:pack_proxy),
+							unpacker: self.method(:unpack_proxy),
+						)
 
 					self.register_type(0x11, Release, recursive: true,
-						packer: ->(release, packer){release.pack(packer)},
-						unpacker: ->(unpacker){Release.unpack(unpacker)},
-					)
+							packer: ->(release, packer){release.pack(packer)},
+							unpacker: ->(unpacker){Release.unpack(unpacker)},
+						)
 
 					self.register_type(0x20, Symbol)
-					self.register_type(0x21, Exception,
-						packer: self.method(:pack_exception),
-						unpacker: self.method(:unpack_exception),
-						recursive: true,
-					)
+					self.register_type(0x21, Exception, recursive: true,
+							packer: self.method(:pack_exception),
+							unpacker: self.method(:unpack_exception),
+						)
 
 					self.register_type(0x22, Class,
-						packer: ->(klass){klass.name},
-						unpacker: ->(name){Object.const_get(name)},
-					)
+							packer: ->(klass){klass.name},
+							unpacker: ->(name){Object.const_get(name)},
+						)
+					
+					reference_packer = self.method(:pack_reference)
+					reference_unpacker = self.method(:unpack_reference)
 
 					# Serialize objects into proxies:
 					reference_types&.each_with_index do |klass, index|
-						self.register_type(0x30 + index, klass,
-							packer: connection.method(:proxy_name),
-							unpacker: connection.method(:[]),
-						)
+						self.register_type(0x30 + index, klass, recursive: true,
+								packer: reference_packer,
+								unpacker: reference_unpacker,
+							)
 					end
 				end
 
+				# Pack a proxy into a MessagePack packer.
+				#
+				# Validates that the proxy is for this connection and serializes the proxy name.
+				# Multi-hop proxy forwarding is not supported, so proxies can only be serialized
+				# from the same connection they were created for (round-trip scenarios).
+				#
+				# @parameter proxy [Proxy] The proxy to serialize.
+				# @parameter packer [MessagePack::Packer] The packer to write to.
+				# @raises [ArgumentError] If the proxy is from a different connection (multi-hop forwarding not supported).
+				def pack_proxy(proxy, packer)
+					# Check if the proxy is for this connection:
+					if proxy.__connection__ != @connection
+						proxy = @connection.proxy(proxy)
+					end
+					
+					packer.write(proxy.__name__)
+				end
+				
+				# Unpack a proxy from a MessagePack unpacker.
+				#
+				# When deserializing a proxy:
+				# - If the object is bound locally, return the actual object (round-trip scenario)
+				# - If the object is not found locally, create a proxy pointing to this connection
+				#   (the proxy was forwarded from another connection and should point back to the sender)
+				#
+				# @parameter unpacker [MessagePack::Unpacker] The unpacker to read from.
+				# @returns [Object | Proxy] The actual object if bound locally, or a proxy pointing to this connection.
+				def unpack_proxy(unpacker)
+					@connection.proxy_object(unpacker.read)
+				end
+				
 				# Pack an exception into a MessagePack packer.
 				# @parameter exception [Exception] The exception to pack.
 				# @parameter packer [MessagePack::Packer] The packer to write to.
@@ -98,6 +135,14 @@ def unpack_exception(unpacker)
 
 					return exception
 				end
+				
+				def pack_reference(object, packer)
+					packer.write(@connection.proxy_name(object))
+				end
+				
+				def unpack_reference(unpacker)
+					@connection.proxy_object(unpacker.read)
+				end
 			end
 		end
 	end

releases.md

@@ -1,5 +1,9 @@
 # Releases
 
+## Unreleased
+
+  - Add support for multi-hop proxying.
+
 ## v0.2.0
 
   - Fix handling of temporary objects.

test/async/bus/client.rb

@@ -44,7 +44,7 @@ def initialize(endpoint, connected_count)
 				end
 			end
 
-			client_task = client_instance.run
+			client_task = Async {client_instance.run}
 
 			# Wait for initial connection
 			reactor.sleep(0.01)
@@ -87,7 +87,7 @@ def initialize(endpoint, connected_count, connection_count)
 				end
 			end
 
-			client_task = client_instance.run
+			client_task = Async {client_instance.run}
 
 			# Wait for initial connection
 			reactor.sleep(0.02)
@@ -130,7 +130,7 @@ def initialize(endpoint, state)
 				end
 			end
 
-			client_task = client_instance.run
+			client_task = Async {client_instance.run}
 
 			# Wait for initial connection
 			reactor.sleep(0.01)
@@ -172,7 +172,7 @@ def initialize(endpoint, error_count)
 				end
 			end
 
-			client_task = client_instance.run
+			client_task = Async {client_instance.run}
 
 			# Wait for reconnection after initial failure
 			# The first attempt fails, sleeps (rand, so 0-1 seconds), then retries