Class: Async::GRPC::XDS::DiscoveryClient

Inherits:
Object
  • Object
show all
Includes:
ADSStream::Delegate
Defined in:
lib/async/grpc/xds/discovery_client.rb

Overview

Client for xDS APIs (ADS or individual APIs) Implements Aggregated Discovery Service (ADS) protocol Acts as delegate for ADSStream, receiving discovery_response events

Constant Summary collapse

LISTENER_TYPE =

xDS API type URLs (v3 API)

"type.googleapis.com/envoy.config.listener.v3.Listener"
ROUTE_TYPE = 
"type.googleapis.com/envoy.config.route.v3.RouteConfiguration"
CLUSTER_TYPE = 
"type.googleapis.com/envoy.config.cluster.v3.Cluster"
ENDPOINT_TYPE = 
"type.googleapis.com/envoy.config.endpoint.v3.ClusterLoadAssignment"
SECRET_TYPE = 
"type.googleapis.com/envoy.extensions.transport_sockets.tls.v3.Secret"

Instance Method Summary collapse

Constructor Details

#initialize(server_config, node: nil) ⇒ DiscoveryClient

Initialize xDS discovery client



37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
 
# File 'lib/async/grpc/xds/discovery_client.rb', line 37

def initialize(server_config, node: nil)
	@server_uri = server_config[:server_uri]
	@channel_creds = server_config[:channel_creds]
	@server_features = server_config[:server_features] || []
	@node_info = node || build_node_info
	@node = build_node_proto(@node_info)
	@grpc_client = nil
	@versions = {}  # Track version_info per type_url
	@nonces = {}     # Track nonces per type_url
	@mutex = Mutex.new
	@subscriptions = {}  # Track subscriptions by type_url
	@stream_task = nil
	@ads_stream = nil  # ADSStream instance when connected (owns stream state)
	@stream_ready_promise = nil  # Resolved when stream_opened runs
end

Instance Method Details

#close ⇒ Object

Close xDS discovery client



88
89
90
91
92
93
94
95
96
97
98
 
# File 'lib/async/grpc/xds/discovery_client.rb', line 88

def close
	@mutex.synchronize do
		@stream_task&.stop
		@grpc_client&.close
		@grpc_client = nil
		@subscriptions.clear
		@stream_task = nil
		@ads_stream = nil
		@stream_ready_promise = nil
	end
end

#discovery_response(response, stream) ⇒ Object

Process a discovery response received by an ADS stream.



171
172
173
 
# File 'lib/async/grpc/xds/discovery_client.rb', line 171

def discovery_response(response, stream)
	process_response(response, stream)
end

#stream_closed(stream) ⇒ Object

Record that an ADS stream has closed.



164
165
166
 
# File 'lib/async/grpc/xds/discovery_client.rb', line 164

def stream_closed(stream)
	@mutex.synchronize{@ads_stream = nil}
end

#stream_opened(stream) ⇒ Object

Record that an ADS stream has opened and unblock pending subscriptions.



157
158
159
160
 
# File 'lib/async/grpc/xds/discovery_client.rb', line 157

def stream_opened(stream)
	@mutex.synchronize{@ads_stream = stream}
	@stream_ready_promise&.resolve(stream)
end

#subscribe(type_url, resource_names, &block) ⇒ Object

Subscribe to resource type using ADS (Aggregated Discovery Service - single stream for all types)



59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
 
# File 'lib/async/grpc/xds/discovery_client.rb', line 59

def subscribe(type_url, resource_names, &block)
	# Store subscription callback
	@mutex.synchronize do
		@subscriptions[type_url] = {
			resource_names: resource_names,
			callback: block
		}
	end
	
	# Ensure ADS stream is running
	ensure_stream_running
	
	# Wait for stream to be ready (event-driven, no polling)
	promise = @stream_ready_promise
	if promise && !promise.completed?
		begin
			promise.wait(timeout: 5)
		rescue Async::TimeoutError
			# Stream didn't open in time; send_discovery_request will no-op if @ads_stream is nil
		end
	end
	
	send_discovery_request(type_url, resource_names) if @ads_stream
	
	# Return the stream task (already running)
	@stream_task
end