`class Registry < Async::Bus::Controller`

Inherits from: Async::Bus::Controller

Maintains live backends and a global queue of available processing slots.

Definitions

`def initialize(exchange_limit: 8, backend_factory: nil)`

Initialize an endpoint registry.

Signature

parameter　exchange_limit　Integer: The maximum outstanding responses per backend.
parameter　backend_factory　Proc | Nil: An optional backend construction strategy.

Implementation

def initialize(exchange_limit: 8, backend_factory: nil)
	@exchange_limit = exchange_limit
	@backend_factory = backend_factory || self.method(:make_backend)
	
	@guard = Thread::Mutex.new
	@backends = {}
	@available = Async::Queue.new
	@closed = false
end

`def replace(descriptions)`

Replace the complete endpoint set.

Signature

parameter　descriptions　Array(Endpoint | Hash): The desired endpoints.
returns　Integer: The resulting endpoint count.

Implementation

def replace(descriptions)
	endpoints = descriptions.map{|description| Endpoint.coerce(description)}
	names = endpoints.map(&:name)
	
	current_names = @guard.synchronize{@backends.keys}
	self.update(endpoints, current_names - names)
end

`def update(upserted, removed)`

Apply endpoint additions, replacements, and removals.

Signature

parameter　upserted　Array(Endpoint | Hash): Endpoints to add or replace.
parameter　removed　Array(String): Endpoint names to remove.
returns　Integer: The resulting endpoint count.

Implementation

def update(upserted, removed)
	retired = []
	started = []
	
	@guard.synchronize do
		raise IOError, "Registry is closed!" if @closed
		
		removed.each do |name|
			if backend = @backends.delete(name.to_s)
				retired << backend
			end
		end
		
		upserted.each do |description|
			endpoint = Endpoint.coerce(description)
			current = @backends[endpoint.name]
			
			next if current&.endpoint == endpoint
			
			retired << current if current
			backend = @backend_factory.call(endpoint, @exchange_limit, self.method(:offer))
			@backends[endpoint.name] = backend
			started << backend
		end
	end
	
	retired.each(&:retire)
	started.each(&:start)
	@available.enqueue(WAKE) unless retired.empty?
	
	self.size
end

`def acquire`

Acquire the next backend with processing capacity.

Signature

returns　Backend | Nil: An admitted backend, or nil if no endpoints exist.

Implementation

def acquire
	loop do
		return nil if self.empty?
		
		candidate = @available.dequeue
		return nil unless candidate
		next if candidate.equal?(WAKE)
		
		return candidate if candidate.reserve
	end
end

`def size`

Signature

returns　Integer: The number of active endpoints.

Implementation

def size
	@guard.synchronize{@backends.size}
end

`def empty?`

Signature

returns　Boolean: Whether no active endpoints are registered.

Implementation

def empty?
	self.size.zero?
end

`def names`

Signature

returns　Array(String): The active endpoint names.

Implementation

def names
	@guard.synchronize{@backends.keys.sort}
end

`def [](name)`

Find an active backend by name.

Signature

parameter　name　String: The endpoint name.
returns　Backend | Nil: The active backend.

Implementation

def [](name)
	@guard.synchronize{@backends[name.to_s]}
end

`def close`

Close the registry and retire all backends.

Implementation

def close
	backends = @guard.synchronize do
		next [] if @closed
		
		@closed = true
		@backends.values.tap{@backends = {}}
	end
	
	@available.close
	backends.each(&:retire)
end