ads

Documentation

Overview

Package ads provides the implementation of an ADS (Aggregated Discovery Service) stream for the xDS client.

Index

• type DataAndErrTuple
• type ResourceWatchState
• type Response
• type StreamEventHandler
• type StreamImpl
  • func NewStreamImpl(opts StreamOpts) *StreamImpl
  • func (s *StreamImpl) ResourceWatchStateForTesting(typ xdsresource.Type, resourceName string) (ResourceWatchState, error)
  • func (s *StreamImpl) Stop()
  • func (s *StreamImpl) Subscribe(typ xdsresource.Type, name string)
  • func (s *StreamImpl) TriggerResourceNotFoundForTesting(typ xdsresource.Type, resourceName string)
  • func (s *StreamImpl) Unsubscribe(typ xdsresource.Type, name string)
• type StreamOpts
• type WatchState

Types

type DataAndErrTuple

type DataAndErrTuple struct {
	Resource xdsresource.ResourceData
	Err      error
}

DataAndErrTuple is a struct that holds a resource and an error. It is used to return a resource and any associated error from a function.

type ResourceWatchState

type ResourceWatchState struct {
	State       WatchState  // Watch state of the resource.
	ExpiryTimer *time.Timer // Timer for the expiry of the watch.
}

ResourceWatchState is the state corresponding to a resource being watched.

type Response

type Response struct {
	TypeURL   string
	Version   string
	Resources []*anypb.Any
}

Response represents a response received on the ADS stream. It contains the type URL, version, and resources for the response.

type StreamEventHandler

type StreamEventHandler interface {
	OnADSStreamError(error)                           // Called when the ADS stream breaks.
	OnADSWatchExpiry(xdsresource.Type, string)        // Called when the watch timer expires for a resource.
	OnADSResponse(Response, func()) ([]string, error) // Called when a response is received on the ADS stream.
}

StreamEventHandler is an interface that defines the callbacks for events that occur on the ADS stream. Methods on this interface may be invoked concurrently and implementations need to handle them in a thread-safe manner.

type StreamImpl

type StreamImpl struct {
	// contains filtered or unexported fields
}

StreamImpl provides the functionality associated with an ADS (Aggregated Discovery Service) stream on the client side. It manages the lifecycle of the ADS stream, including creating the stream, sending requests, and handling responses. It also handles flow control and retries for the stream.

func NewStreamImpl

func NewStreamImpl(opts StreamOpts) *StreamImpl

NewStreamImpl initializes a new StreamImpl instance using the given parameters. It also launches goroutines responsible for managing reads and writes for messages of the underlying stream.

func (*StreamImpl) ResourceWatchStateForTesting

func (s *StreamImpl) ResourceWatchStateForTesting(typ xdsresource.Type, resourceName string) (ResourceWatchState, error)

ResourceWatchStateForTesting returns the ResourceWatchState for the given resource type and name. This is intended for testing purposes only, to inspect the internal state of the ADS stream.

func (*StreamImpl) Stop

func (s *StreamImpl) Stop()

Stop blocks until the stream is closed and all spawned goroutines exit.

func (*StreamImpl) Subscribe

func (s *StreamImpl) Subscribe(typ xdsresource.Type, name string)

Subscribe subscribes to the given resource. It is assumed that multiple subscriptions for the same resource is deduped at the caller. A discovery request is sent out on the underlying stream for the resource type when there is sufficient flow control quota.

func (*StreamImpl) TriggerResourceNotFoundForTesting

func (s *StreamImpl) TriggerResourceNotFoundForTesting(typ xdsresource.Type, resourceName string)

TriggerResourceNotFoundForTesting triggers a resource not found event for the given resource type and name. This is intended for testing purposes only, to simulate a resource not found scenario.

func (*StreamImpl) Unsubscribe

func (s *StreamImpl) Unsubscribe(typ xdsresource.Type, name string)

Unsubscribe cancels the subscription to the given resource. It is a no-op if the given resource does not exist. The watch expiry timer associated with the resource is stopped if one is active. A discovery request is sent out on the stream for the resource type when there is sufficient flow control quota.

type StreamOpts

type StreamOpts struct {
	Transport          transport.Transport     // xDS transport to create the stream on.
	EventHandler       StreamEventHandler      // Callbacks for stream events.
	Backoff            func(int) time.Duration // Backoff for retries, after stream failures.
	NodeProto          *v3corepb.Node          // Node proto to identify the gRPC application.
	WatchExpiryTimeout time.Duration           // Resource watch expiry timeout.
	LogPrefix          string                  // Prefix to be used for log messages.
}

StreamOpts contains the options for creating a new ADS Stream.

type WatchState

type WatchState int

WatchState is a enum that describes the watch state of a particular resource.

const (
	// ResourceWatchStateStarted is the state where a watch for a resource was
	// started, but a request asking for that resource is yet to be sent to the
	// management server.
	ResourceWatchStateStarted WatchState = iota
	// ResourceWatchStateRequested is the state when a request has been sent for
	// the resource being watched.
	ResourceWatchStateRequested
	// ResourceWatchStateReceived is the state when a response has been received
	// for the resource being watched.
	ResourceWatchStateReceived
	// ResourceWatchStateTimeout is the state when the watch timer associated
	// with the resource expired because no response was received.
	ResourceWatchStateTimeout
)