Use proper types for protocol.

Author: ioquatix

bake/async/bus.rb

@@ -5,9 +5,10 @@ def server
 
 	Async do
 		server = Async::Bus::Server.new
+		things = Array.new
 
 		server.accept do |connection|
-			binding.irb
+			connection[:things] = things
 		end
 	end
 end

lib/async/bus/protocol/connection.rb

@@ -41,6 +41,20 @@ def initialize(peer, id)
 					@finalized = ::Thread::Queue.new
 				end
 
+				def flush
+					@packer.flush
+				end
+				
+				def write(message)
+					# $stderr.puts "Writing: #{message.inspect}"
+					@packer.write(message)
+					@packer.flush
+				end
+				
+				def inspect
+					"#<#{self.class} #{@objects.size} objects>"
+				end
+				
 				attr :objects
 				attr :proxies
 
@@ -79,6 +93,10 @@ def bind(name, object)
 					proc{@finalized << name}
 				end
 
+				def []=(name, object)
+					@objects[name] = object
+				end
+				
 				def [](name)
 					unless proxy = @proxies[name]
 						proxy = Proxy.new(self, name)
@@ -92,47 +110,47 @@ def [](name)
 
 				def invoke(name, arguments, options = {}, &block)
 					id = self.next_id
+					# $stderr.puts "-> Invoking: #{name} #{arguments.inspect} #{options.inspect}", caller
 
 					transaction = Transaction.new(self, id)
 					@transactions[id] = transaction
 
 					transaction.invoke(name, arguments, options, &block)
 				ensure
 					transaction&.close
+					# $stderr.puts "<- Invoked: #{name}"
 				end
 
 				def run
 					finalizer_task = Async do
 						while name = @finalized.pop
-							@packer.write([:release, name])
+							self.write(Release.new(name))
 						end
 					end
 
 					@unpacker.each do |message|
-						id = message.shift
+						# $stderr.puts "Message received: #{message.inspect}"
 
-						if id == :release
-							name = message.shift
-							@objects.delete(name) if name.is_a?(String)
-						elsif transaction = @transactions[id]
-							transaction.received.enqueue(message)
-						elsif message.first == :invoke
-							message.shift
-							
-							transaction = Transaction.new(self, id)
-							@transactions[id] = transaction
+						case message
+						when Release
+							@objects.delete(message.name)
+						when Invoke
+							transaction = Transaction.new(self, message.id)
+							@transactions[message.id] = transaction
 
-							name = message.shift
-							object = @objects[name]
+							object = @objects[message.name]
 
 							Async do
-								transaction.accept(object, *message)
+								# $stderr.puts "-> Accepting: #{message.name} #{message.arguments.inspect} #{message.options.inspect}"
+								transaction.accept(object, message.arguments, message.options, message.block_given)
 							ensure
+								# $stderr.puts "<- Accepted: #{message.name}"
 								# This will also delete the transaction from @transactions:
 								transaction.close
 							end
 						else
-							raise "Out of order message: #{message}"
+							transaction = @transactions[message.id]
+							transaction.received.enqueue(message)
 						end
 					end
 				ensure

lib/async/bus/protocol/proxy.rb

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

lib/async/bus/protocol/transaction.rb

@@ -22,16 +22,17 @@ def initialize(connection, id)
 
 				def read
 					if @received.empty?
-						@connection.packer.flush
+						@connection.flush
 					end
 
 					@received.dequeue
 				end
 
-				def write(*arguments)
+				def write(message)
+					# $stderr.puts "Transaction Writing: #{message.inspect}"
+					
 					if @connection
-						@connection.packer.write([id, *arguments])
-						@connection.packer.flush
+						@connection.write(message)
 					else
 						raise RuntimeError, "Transaction is closed!"
 					end
@@ -50,23 +51,21 @@ def close
 				def invoke(name, arguments, options, &block)
 					Console.debug(self) {[name, arguments, options, block]}
 
-					self.write(:invoke, name, arguments, options, block_given?)
+					self.write(Invoke.new(@id, name, arguments, options, block_given?))
 
 					while response = self.read
-						what, result = response
-						
-						case what
-						when :error
-							raise(result)
-						when :return
-							return(result)
-						when :yield
+						case response
+						when Return
+							return response.result
+						when Yield
 							begin
-								result = yield(*result)
-								self.write(:next, result)
+								result = yield(*response.result)
+								self.write(Next.new(@id, result))
 							rescue => error
-								self.write(:error, error)
+								self.write(Error.new(@id, error))
 							end
+						when Error
+							raise(response.result)
 						end
 					end
 
@@ -75,30 +74,31 @@ def invoke(name, arguments, options, &block)
 				end
 
 				# Accept a remote procedure invokation.
-				def accept(object, arguments, options, block)
-					if block
+				def accept(object, arguments, options, block_given)
+					if block_given
 						result = object.public_send(*arguments, **options) do |*yield_arguments|
-							self.write(:yield, yield_arguments)
-							what, result = self.read
+							self.write(Yield.new(@id, yield_arguments))
+							
+							response = self.read
 
-							case what
-							when :next
-								result
-							when :close
-								return
-							when :error
-								raise(result)
+							case response
+							when Next
+								response.result
+							when Error
+								raise(response.result)
+							when Close
+								break
 							end
 						end
 					else
 						result = object.public_send(*arguments, **options)
 					end
 
-					self.write(:return, result)
+					self.write(Return.new(@id, result))
 				rescue UncaughtThrowError => error
-					self.write(:throw, error.tag)
+					self.write(Throw.new(@id, error.tag))
 				rescue => error
-					self.write(:error, error)
+					self.write(Error.new(@id, error))
 				# ensure
 				# 	self.write(:close)
 				end