`class ControlPlane`

Maintains xDS resource snapshots and notifies discovery streams when resources change.

Definitions

`def initialize(identifier: "async-grpc-xds")`

Initialize an empty control plane.

Signature

parameter　identifier　String: The identifier reported in discovery responses.

Implementation

def initialize(identifier: "async-grpc-xds")
	@identifier = identifier
	@resources = Hash.new{|hash, type_url| hash[type_url] = {}}
	@versions = Hash.new(0)
	@streams = Set.new.compare_by_identity
	@mutex = Mutex.new
end

`def update_cluster(name, resource = nil, **options)`

Add or replace a cluster resource.

Signature

parameter　name　String: The cluster name.
parameter　resource　Envoy::Config::Cluster::V3::Cluster | Nil: An existing cluster resource, or nil to build one from options.
parameter　options　Hash: Options forwarded to Async::GRPC::XDS::Cluster#build.

Implementation

def update_cluster(name, resource = nil, **options)
	resource ||= Cluster.build(name, **options)
	update_resource(CLUSTER_TYPE, name.to_s, resource)
end

`def update_endpoints(cluster_name, endpoints)`

Add or replace the endpoint assignment for a cluster.

Signature

parameter　cluster_name　String: The cluster name.
parameter　endpoints　Array(Hash): The normalized endpoint states.

Implementation

def update_endpoints(cluster_name, endpoints)
	update_resource(
		ENDPOINT_TYPE,
		cluster_name.to_s,
		Endpoint.build(cluster_name, endpoints)
	)
end

`def remove_cluster(name)`

Remove a cluster resource.

Signature

parameter　name　String: The cluster name.

Implementation

def remove_cluster(name)
	remove_resource(CLUSTER_TYPE, name.to_s)
end

`def remove_endpoints(cluster_name)`

Remove the endpoint assignment for a cluster.

Signature

parameter　cluster_name　String: The cluster name.

Implementation

def remove_endpoints(cluster_name)
	remove_resource(ENDPOINT_TYPE, cluster_name.to_s)
end

`def update_resource(type_url, name, resource)`

Add or replace an xDS resource and notify subscribed streams.

Signature

parameter　type_url　String: The xDS resource type URL.
parameter　name　String: The resource name.
parameter　resource　Google::Protobuf::MessageExts: The protobuf resource.

Implementation

def update_resource(type_url, name, resource)
	notify = false
	
	@mutex.synchronize do
		@resources[type_url][name] = resource
		@versions[type_url] += 1
		notify = true
	end
	
	notify_streams(type_url) if notify
end

`def remove_resource(type_url, name)`

Remove an xDS resource and notify subscribed streams.

Signature

parameter　type_url　String: The xDS resource type URL.
parameter　name　String: The resource name.

Implementation

def remove_resource(type_url, name)
	notify = false
	
	@mutex.synchronize do
		if @resources[type_url].delete(name)
			@versions[type_url] += 1
			notify = true
		end
	end
	
	notify_streams(type_url) if notify
end

`def resource_names(type_url)`

Get the available resource names for a type.

Signature

parameter　type_url　String: The xDS resource type URL.
returns　Array(String): The resource names.

Implementation

def resource_names(type_url)
	@mutex.synchronize do
		@resources[type_url].keys
	end
end

`def resources(type_url, names = nil)`

Get resources of a given type.

Signature

parameter　type_url　String: The xDS resource type URL.
parameter　names　Array(String) | Nil: The requested resource names, or nil for all resources.
returns　Array(Google::Protobuf::MessageExts): The matching resources.

Implementation

def resources(type_url, names = nil)
	@mutex.synchronize do
		resources = @resources[type_url]
		
		if names && names.any?
			names.filter_map{|name| resources[name]}
		else
			resources.values
		end
	end
end

`def version(type_url)`

Get the current version for a resource type.

Signature

parameter　type_url　String: The xDS resource type URL.
returns　String: The monotonically increasing version.

Implementation

def version(type_url)
	@mutex.synchronize do
		@versions[type_url].to_s
	end
end

`def response(type_url, names = nil)`

Build a discovery response for a resource type.

Signature

parameter　type_url　String: The xDS resource type URL.
parameter　names　Array(String) | Nil: The requested resource names, or nil for all resources.
returns　Envoy::Service::Discovery::V3::DiscoveryResponse: The current discovery response.

Implementation

def response(type_url, names = nil)
	resources = self.resources(type_url, names)
	version = self.version(type_url)
	
	Envoy::Service::Discovery::V3::DiscoveryResponse.new(
		version_info: version,
		resources: resources.map{|resource| Google::Protobuf::Any.pack(resource)},
		type_url: type_url,
		nonce: "#{type_url}:#{version}:#{SecureRandom.hex(8)}",
		control_plane: Envoy::Config::Core::V3::ControlPlane.new(identifier: @identifier)
	)
end

`def register_stream(stream)`

Signature

parameter　stream　Stream: The stream to register.

Implementation

def register_stream(stream)
	@mutex.synchronize do
		@streams.add(stream)
	end
end

`def remove_stream(stream)`

Remove a registered stream.

Signature

parameter　stream　Stream: The stream to remove.

Implementation

def remove_stream(stream)
	@mutex.synchronize do
		@streams.delete(stream)
	end
end