Class: Async::GRPC::XDS::Service

Inherits:
Service
  • Object
show all
Defined in:
lib/async/grpc/xds/service.rb

Overview

Serves Envoy Aggregated Discovery Service requests from a ControlPlane.

Defined Under Namespace

Classes: Stream

Constant Summary collapse

SERVICE_NAME = 
"envoy.service.discovery.v3.AggregatedDiscoveryService"

Instance Method Summary collapse

Constructor Details

#initialize(control_plane) ⇒ Service

Initialize an Aggregated Discovery Service.



26
27
28
29
 
# File 'lib/async/grpc/xds/service.rb', line 26

def initialize(control_plane)
	super(Envoy::Service::Discovery::V3::AggregatedDiscoveryService, SERVICE_NAME)
	@control_plane = control_plane
end

Instance Method Details

#delta_aggregated_resources(input, output, call) ⇒ Object

Reject a delta Aggregated Discovery Service stream, which is not supported.

Raises:

  • (Protocol::GRPC::Error) 


63
64
65
66
67
68
 
# File 'lib/async/grpc/xds/service.rb', line 63

def delta_aggregated_resources(input, output, call)
	raise Protocol::GRPC::Error.new(
		Protocol::GRPC::Status::UNIMPLEMENTED,
		"Delta xDS is not implemented."
	)
end

#stream_aggregated_resources(input, output, call) ⇒ Object

Serve a state-of-the-world Aggregated Discovery Service stream.



36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
 
# File 'lib/async/grpc/xds/service.rb', line 36

def stream_aggregated_resources(input, output, call)
	stream = Stream.new(@control_plane, output)
	@control_plane.register_stream(stream)
	
	reader = Async do
		input.each do |request|
			stream.request(request)
		end
	end
	
	writer = Async do
		stream.run
	end
	
	reader.wait
ensure
	stream&.close
	reader&.stop
	writer&.stop
	@control_plane.remove_stream(stream) if stream
end