Explicit proxying.

Author: samuel-williams-shopify

lib/async/bus.rb

@@ -4,3 +4,4 @@
 # Copyright, 2021-2025, by Samuel Williams.
 
 require_relative "bus/version"
+require_relative "bus/controller"

lib/async/bus/client.rb

@@ -9,14 +9,15 @@
 module Async
 	module Bus
 		class Client
-			def initialize(endpoint = nil)
+			def initialize(endpoint = nil, **options)
 				@endpoint = endpoint || Protocol.local_endpoint
+				@options = options
 			end
 
 			# @parameter persist [Boolean] Whether to keep the connection open indefiniely.
 			def connect(persist = false)
 				@endpoint.connect do |peer|
-					connection = Protocol::Connection.client(peer)
+					connection = Protocol::Connection.client(peer, **@options)
 
 					connection_task = Async do
 						connection.run

lib/async/bus/controller.rb

@@ -0,0 +1,58 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2021-2025, by Samuel Williams.
+
+module Async
+	module Bus
+		# Base class for controller objects designed to be proxied over Async::Bus.
+		#
+		# Controllers provide an explicit API for remote operations, avoiding the
+		# confusion that comes from proxying generic objects like Array or Hash.
+		#
+		# @example Array Controller
+		#   class ArrayController < Async::Bus::Controller
+		#     def initialize(array)
+		#       @array = array
+		#     end
+		#
+		#     def append(*values)
+		#       @array.concat(values)
+		#       self  # Return self for chaining
+		#     end
+		#
+		#     def get(index)
+		#       @array[index]  # Returns value
+		#     end
+		#
+		#     def size
+		#       @array.size
+		#     end
+		#   end
+		#
+		# @example Server Setup
+		#   server.accept do |connection|
+		#     array = []
+		#     controller = ArrayController.new(array)
+		#     connection.bind(:items, controller)
+		#   end
+		#
+		# @example Client Usage
+		#   client.connect do |connection|
+		#     items = connection[:items]  # Returns proxy to controller
+		#     items.append(1, 2, 3)       # Remote call
+		#     expect(items.size).to be == 3
+		#   end
+		#
+		# Controllers are automatically proxied when serialized if registered
+		# as a reference type in the Wrapper:
+		#
+		#   Wrapper.new(connection, reference_types: [Async::Bus::Controller])
+		#
+		# This allows controller methods to return other controllers and have
+		# them automatically proxied.
+		class Controller
+		end
+	end
+end
+

lib/async/bus/protocol/connection.rb

@@ -18,18 +18,18 @@ def self.local_endpoint(path = "bus.ipc")
 			end
 
 			class Connection
-				def self.client(peer)
-					self.new(peer, 1)
+				def self.client(peer, **options)
+					self.new(peer, 1, **options)
 				end
 
-				def self.server(peer)
-					self.new(peer, 2)
+				def self.server(peer, **options)
+					self.new(peer, 2, **options)
 				end
 
-				def initialize(peer, id)
+				def initialize(peer, id, wrapper: Wrapper)
 					@peer = peer
 
-					@wrapper = Wrapper.new(self)
+					@wrapper = wrapper.new(self)
 					@unpacker = @wrapper.unpacker(peer)
 					@packer = @wrapper.packer(peer)
 
@@ -72,23 +72,35 @@ def next_id
 
 				# Bind a local object to a name, such that it could be accessed remotely.
 				#
-				# @returns [String] The (unique) name of the object.
+				# @returns [Proxy] A proxy instance for the bound object.
+				def bind(name, object)
+					@objects[name] = object
+					return self[name]
+				end
+				
+				# Generate a proxy name for an object and bind it.
+				#
+				# @returns [Proxy] A proxy instance for the bound object.
 				def proxy(object)
 					name = "<#{object.class}@#{next_id.to_s(16)}>".freeze
 
+					return bind(name, object)
+				end
+				
+				# Generate a proxy name for an object and bind it, returning just the name.
+				# Used for serialization when you need the name string, not a Proxy instance.
+				#
+				# @returns [String] The name of the bound object.
+				def proxy_name(object)
+					name = "<#{object.class}@#{next_id.to_s(16)}>".freeze
 					bind(name, object)
-					
 					return name
 				end
 
 				def object(name)
 					@objects[name]
 				end
 
-				def bind(name, object)
-					@objects[name] = object
-				end
-				
 				private def finalize(name)
 					proc{@finalized << name}
 				end
@@ -102,7 +114,7 @@ def [](name)
 						proxy = Proxy.new(self, name)
 						@proxies[name] = proxy
 
-						ObjectSpace.define_finalizer(proxy, finalize(name))
+						::ObjectSpace.define_finalizer(proxy, finalize(name))
 					end
 
 					return proxy
@@ -150,7 +162,7 @@ def run
 							end
 						else
 							transaction = @transactions[message.id]
-							transaction.received.enqueue(message)
+							transaction.received.push(message)
 						end
 					end
 				ensure

lib/async/bus/protocol/invoke.rb

@@ -0,0 +1,58 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2025, by Samuel Williams.
+
+require "msgpack"
+require_relative "proxy"
+
+module Async
+	module Bus
+		module Protocol
+			# Represents a method invocation.
+			class Invoke
+				def initialize(id, name, arguments, options, block_given)
+					@id = id
+					@name = name
+					@arguments = arguments
+					@options = options
+					@block_given = block_given
+				end
+				
+				attr :id
+				attr :name
+				attr :arguments
+				attr :options
+				attr :block_given
+				
+				def pack(packer)
+					packer.write(@id)
+					packer.write(@name)
+					
+					packer.write(@arguments.size)
+					@arguments.each do |argument|
+						packer.write(argument)
+					end
+					
+					packer.write(@options.size)
+					@options.each do |key, value|
+						packer.write(key)
+						packer.write(value)
+					end
+					
+					packer.write(@block_given)
+				end
+				
+				def self.unpack(unpacker)
+					id = unpacker.read
+					name = unpacker.read
+					arguments = Array.new(unpacker.read) {unpacker.read}
+					options = Array.new(unpacker.read) {[unpacker.read, unpacker.read]}.to_h
+					block_given = unpacker.read
+					
+					return self.new(id, name, arguments, options, block_given)
+				end
+			end
+		end
+	end
+end

lib/async/bus/protocol/proxy.rb

@@ -48,7 +48,7 @@ def respond_to_missing?(name, include_all = false)
 				end
 
 				def inspect
-					"#<proxy #{@name}>"
+					"#<proxy #{@name}: #{@connection.invoke(@name, [:inspect])}>"
 				end
 			end
 		end

lib/async/bus/protocol/release.rb

@@ -0,0 +1,29 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2025, by Samuel Williams.
+
+module Async
+	module Bus
+		module Protocol
+			# Represents a named object that has been released (no longer available).
+			class Release
+				def initialize(name)
+					@name = name
+				end
+				
+				attr :name
+				
+				def pack(packer)
+					packer.write(@name)
+				end
+				
+				def self.unpack(unpacker)
+					name = unpacker.read
+					
+					return self.new(name)
+				end
+			end
+		end
+	end
+end

lib/async/bus/protocol/response.rb

@@ -0,0 +1,39 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2025, by Samuel Williams.
+
+module Async
+	module Bus
+		module Protocol
+			class Response
+				def initialize(id, result)
+					@id = id
+					@result = result
+				end
+				
+				attr :id
+				attr :result
+				
+				def pack(packer)
+					packer.write(@id)
+					packer.write(@result)
+				end
+				
+				def self.unpack(unpacker)
+					id = unpacker.read
+					result = unpacker.read
+					
+					return self.new(id, result)
+				end
+			end
+			
+			Return = Class.new(Response)
+			Yield = Class.new(Response)
+			Error = Class.new(Response)
+			Next = Class.new(Response)
+			Throw = Class.new(Response)
+			Close = Class.new(Response)
+		end
+	end
+end

lib/async/bus/protocol/transaction.rb

@@ -9,11 +9,13 @@ module Async
 	module Bus
 		module Protocol
 			class Transaction
-				def initialize(connection, id)
+				def initialize(connection, id, timeout: nil)
 					@connection = connection
 					@id = id
+
+					@timeout = timeout
 
-					@received = Async::Queue.new
+					@received = Thread::Queue.new
 					@accept = nil
 				end
 
@@ -25,12 +27,10 @@ def read
 						@connection.flush
 					end
 
-					@received.dequeue
+					@received.pop(timeout: @timeout)
 				end
 
 				def write(message)
-					# $stderr.puts "Transaction Writing: #{message.inspect}"
-					
 					if @connection
 						@connection.write(message)
 					else
@@ -41,7 +41,7 @@ def write(message)
 				def close
 					if connection = @connection
 						@connection = nil
-						@received.enqueue(nil)
+						@received.close
 
 						connection.transactions.delete(@id)
 					end