Class: Async::GRPC::XDS::DiscoveryService

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

Overview

Shared implementation for state-of-the-world xDS discovery services.

Instance Method Summary collapse

Constructor Details

#initialize(interface, service_name, control_plane, resource_type: nil) ⇒ DiscoveryService

Initialize a discovery service.



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

def initialize(interface, service_name, control_plane, resource_type: nil)
	super(interface, service_name)
	
	@control_plane = control_plane
	@resource_type = resource_type
end

Instance Method Details

#delta_resources ⇒ Object

Reject a delta discovery stream, which is not supported.

Raises:

  • (Protocol::GRPC::Error) 


59
60
61
62
63
64
 
# File 'lib/async/grpc/xds/discovery_service.rb', line 59

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

#stream_resources(input, output) ⇒ Object

Serve a state-of-the-world discovery stream.



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

def stream_resources(input, output)
	stream = Stream.new(@control_plane, output, resource_type: @resource_type)
	@control_plane.register_stream(stream)
	
	reader = Async::Task.current.async do
		input.each do |request|
			stream.request(request)
		end
	end
	
	writer = Async::Task.current.async do
		stream.run
	end
	
	reader.wait
ensure
	stream&.close
	reader&.stop
	writer&.stop
	@control_plane.remove_stream(stream) if stream
end