Tidy up transaction and connection handling.

Author: ioquatix

lib/async/bus/protocol/connection.rb

@@ -56,6 +56,9 @@ def next_id
 
 				attr :transactions
 
+				# Bind a local object to a name, such that it could be accessed remotely.
+				#
+				# @returns [String] The (unique) name of the object.
 				def proxy(object)
 					name = "<#{object.class}@#{next_id.to_s(16)}>".freeze
 
@@ -94,6 +97,8 @@ def invoke(name, arguments, options = {}, &block)
 					@transactions[id] = transaction
 
 					transaction.invoke(name, arguments, options, &block)
+				ensure
+					transaction&.close
 				end
 
 				def run
@@ -123,6 +128,7 @@ def run
 							Async do
 								transaction.accept(object, *message)
 							ensure
+								# This will also delete the transaction from @transactions:
 								transaction.close
 							end
 						else

lib/async/bus/protocol/transaction.rb

@@ -29,24 +29,26 @@ def read
 				end
 
 				def write(*arguments)
-					@connection.packer.write([id, *arguments])
-					@connection.packer.flush
+					if @connection
+						@connection.packer.write([id, *arguments])
+						@connection.packer.flush
+					else
+						raise RuntimeError, "Transaction is closed!"
+					end
 				end
 
 				def close
-					if @connection
-						@received.enqueue(nil)
-						
-						connection = @connection
+					if connection = @connection
 						@connection = nil
+						@received.enqueue(nil)
 
 						connection.transactions.delete(@id)
 					end
 				end
 
 				# Invoke a remote procedure.
 				def invoke(name, arguments, options, &block)
-					Console.logger.debug(self) {[name, arguments, options, block]}
+					Console.debug(self) {[name, arguments, options, block]}
 
 					self.write(:invoke, name, arguments, options, block_given?)
 
@@ -68,8 +70,8 @@ def invoke(name, arguments, options, &block)
 						end
 					end
 
-					# ensure
-					# 	self.write(:close)
+				# ensure
+				# 	self.write(:close)
 				end
 
 				# Accept a remote procedure invokation.
@@ -97,8 +99,8 @@ def accept(object, arguments, options, block)
 					self.write(:throw, error.tag)
 				rescue => error
 					self.write(:error, error)
-					# ensure
-					# 	self.write(:close)
+				# ensure
+				# 	self.write(:close)
 				end
 			end
 		end

lib/async/bus/server.rb

@@ -11,21 +11,16 @@ module Bus
 		class Server
 			def initialize(endpoint = nil)
 				@endpoint = endpoint || Protocol.local_endpoint
-				@connected = {}
 			end
 
-			attr :connected
-			
 			def accept
 				@endpoint.accept do |peer|
 					connection = Protocol::Connection.server(peer)
-					@connected[peer] = connection
 
 					yield connection
 
 					connection.run
 				ensure
-					connection = @connected.delete(peer)
 					connection&.close
 				end
 			end

test/async/bus/server.rb

@@ -5,6 +5,20 @@
 
 require "async/bus/a_server"
 
+class MyArray
+	def initialize
+		@values = []
+	end
+	
+	def sum
+		@values.sum
+	end
+	
+	def <<(value)
+		@values << value
+	end
+end
+
 describe Async::Bus::Server do
 	include Async::Bus::AServer
 
@@ -116,4 +130,41 @@ def after(error = nil)
 			end
 		end
 	end
+	
+	with "a bound Hash instance" do
+		let(:hash) {Hash.new}
+		
+		def before
+			super
+			
+			@server_task = Async do
+				server.accept do |connection|
+					connection.bind(:hash, hash)
+					connection.bind(:sum_key, proc{|key| hash[key].sum})
+				end
+			end
+		end
+		
+		def after(error = nil)
+			@server_task.stop
+		end
+		
+		it "can assign a local object to the hash" do
+			# We can't use a primitive type here, because it will be serialized and deserialized (e.g. copied), losing the object reference.
+			array = MyArray.new
+			
+			client.connect do |connection|
+				# As an array is a primitve type, we must force it to be a proxy:
+				connection[:hash][:array] = array
+				
+				expect(connection[:sum_key].call(:array)).to be == 0
+				
+				array << 1 << 2 << 3
+				
+				expect(connection[:sum_key].call(:array)).to be == 6
+				
+				expect(connection.transactions).to be(:empty?)
+			end
+		end
+	end
 end