class Registry < Async::Bus::Controller

Inherits from
Async::Bus::Controller

Maintains live backends and notifies the scheduler when capacity changes.

Definitions

def initialize(exchange_limit: 8, permit_limit: 1, backend_factory: nil)

Initialize an endpoint registry.

Signature
parameter　exchange_limit　Integer

The maximum outstanding responses per backend.

parameter　permit_limit　Integer

The maximum active processing permits per backend.

parameter　backend_factory　#call(endpoint, exchange_limit, permit_limit, available) | Nil

An optional backend construction strategy.

Implementation

 
def initialize(exchange_limit: 8, permit_limit: 1, backend_factory: nil)
	@exchange_limit = exchange_limit
	@permit_limit = permit_limit
	@backend_factory = backend_factory || self.method(:make_backend)
	
	@guard = Thread::Mutex.new
	@backends = {}
	@available = nil
	@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, @permit_limit, self.method(:offer))
			@backends[endpoint.name] = backend
			started << backend
		end
	end
	
	retired.each(&:retire)
	started.each(&:start)
	notify_available unless retired.empty?
	
	self.size
end

def backends

Signature
returns　Array(Backend)

A snapshot of active backends.

Implementation

 
def backends
	@guard.synchronize{@backends.values.dup}
end

def on_available(&block)

Register the central scheduler capacity callback.

Implementation

 
def on_available(&block)
	@guard.synchronize{@available = block}
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
	
	backends.each(&:retire)
end