`class Monitor < Async::Bus::Client`

Inherits from: Async::Bus::Client

Publishes a desired endpoint set and subsequent deltas to a Fantail registry.

Definitions

`def initialize(endpoint)`

Initialize an endpoint monitor.

Signature

parameter　endpoint　IO::Endpoint: The Fantail control endpoint.

Implementation

def initialize(endpoint)
	super(endpoint)
	
	@guard = Thread::Mutex.new
	@endpoints = {}
	@revision = 0
	@changes = Async::Queue.new
end

`def replace(descriptions)`

Replace the desired endpoint set without blocking on the control connection.

Signature

parameter　descriptions　Array(Endpoint | Hash): The complete desired endpoint set.
returns　Integer: The local revision number.

Implementation

def replace(descriptions)
	endpoints = descriptions.map{|description| Endpoint.coerce(description).to_h}
	
	revision = @guard.synchronize do
		@endpoints = endpoints.to_h{|description| [description.fetch("name"), description]}
		@revision += 1
	end
	
	@changes.enqueue([revision, :replace, endpoints])
	return revision
end

`def upsert(description)`

Add or replace a desired endpoint without blocking on the control connection.

Signature

parameter　description　Endpoint | Hash: The endpoint to publish.
returns　Integer: The local revision number.

Implementation

def upsert(description)
	endpoint = Endpoint.coerce(description).to_h
	
	revision = @guard.synchronize do
		@endpoints[endpoint.fetch("name")] = endpoint
		@revision += 1
	end
	
	@changes.enqueue([revision, :update, [endpoint], []])
	return revision
end

`def remove(name)`

Remove a desired endpoint without blocking on the control connection.

Signature

parameter　name　String: The endpoint identity to remove.
returns　Integer: The local revision number.

Implementation

def remove(name)
	name = name.to_s
	
	revision = @guard.synchronize do
		@endpoints.delete(name)
		@revision += 1
	end
	
	@changes.enqueue([revision, :update, [], [name]])
	return revision
end

`def endpoints`

Signature

returns　Array(Hash): A snapshot of the desired endpoints.

Implementation

def endpoints
	@guard.synchronize{@endpoints.values.map(&:dup)}
end

`def run(**options)`

Run the persistent publisher, resynchronizing completely after each reconnect.

Signature

parameter　options　Hash: Options forwarded to Async::Bus::Client#run.
returns　Async::Task: The publisher task.

Implementation

def run(**options)
	super(**options) do |connection|
		synchronize(connection[:registry])
	end
end