/*
 *
 * Copyright 2025 gRPC authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */

package xdsclient

import (
	"context"
	"fmt"
	"sync"
	"sync/atomic"
	"time"

	"google.golang.org/grpc/grpclog"
	igrpclog "google.golang.org/grpc/internal/grpclog"
	"google.golang.org/grpc/internal/xds/clients"
	"google.golang.org/grpc/internal/xds/clients/internal/backoff"
	"google.golang.org/grpc/internal/xds/clients/internal/pretty"
	"google.golang.org/grpc/internal/xds/clients/xdsclient/internal/xdsresource"

	"google.golang.org/protobuf/proto"
	"google.golang.org/protobuf/types/known/anypb"

	v3corepb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
	v3discoverypb "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3"
	cpb "google.golang.org/genproto/googleapis/rpc/code"
	statuspb "google.golang.org/genproto/googleapis/rpc/status"
)

const (
	// Any per-RPC level logs which print complete request or response messages
	// should be gated at this verbosity level. Other per-RPC level logs which print
	// terse output should be at `INFO` and verbosity 2.
	perRPCVerbosityLevel = 9
)

// request represents a queued request message to be sent on the ADS stream. It
// contains the type of the resource and the list of resource names to be sent.
type request struct {
	typ           ResourceType
	resourceNames []string
}

// response represents a response received on the ADS stream. It contains the
// type URL, version, and resources for the response.
type response struct {
	typeURL   string
	version   string
	resources []*anypb.Any
}

// 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 dataAndErrTuple struct {
	Resource ResourceData
	Err      error
}

// adsStreamEventHandler 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 adsStreamEventHandler interface {
	onStreamError(error)                           // Called when the ADS stream breaks.
	onWatchExpiry(ResourceType, string)            // Called when the watch timer expires for a resource.
	onResponse(response, func()) ([]string, error) // Called when a response is received on the ADS stream.
}

// state corresponding to a resource type.
type resourceTypeState struct {
	version             string                                     // Last acked version. Should not be reset when the stream breaks.
	nonce               string                                     // Last received nonce. Should be reset when the stream breaks.
	subscribedResources map[string]*xdsresource.ResourceWatchState // Map of subscribed resource names to their state.
}

// adsStreamImpl 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.
type adsStreamImpl struct {
	// The following fields are initialized from arguments passed to the
	// constructor and are read-only afterwards, and hence can be accessed
	// without a mutex.
	transport          clients.Transport       // Transport to use for ADS stream.
	eventHandler       adsStreamEventHandler   // Callbacks into the xdsChannel.
	backoff            func(int) time.Duration // Backoff for retries, after stream failures.
	nodeProto          *v3corepb.Node          // Identifies the gRPC application.
	watchExpiryTimeout time.Duration           // Resource watch expiry timeout
	logger             *igrpclog.PrefixLogger

	// The following fields are initialized in the constructor and are not
	// written to afterwards, and hence can be accessed without a mutex.
	streamCh     chan clients.Stream // New ADS streams are pushed here.
	runnerDoneCh chan struct{}       // Notify completion of runner goroutine.
	cancel       context.CancelFunc  // To cancel the context passed to the runner goroutine.
	fc           *adsFlowControl     // Flow control for ADS stream.
	notifySender chan struct{}       // To notify the sending goroutine of a pending request.

	// Guards access to the below fields (and to the contents of the map).
	mu                sync.Mutex
	resourceTypeState map[ResourceType]*resourceTypeState // Map of resource types to their state.
	pendingRequests   []request                           // Subscriptions and unsubscriptions are pushed here.

	// The following fields are accessed atomically.
	firstRequest       atomic.Bool // False after the first request is sent out.
	firstStreamCreated atomic.Bool // Set to true after the very first ADS stream is created.
	streamEstablished  atomic.Bool // Set to true when an ADS stream is established and a response is received, except for the very first stream which is set to true immediately.
}

// adsStreamOpts contains the options for creating a new ADS Stream.
type adsStreamOpts struct {
	transport          clients.Transport       // xDS transport to create the stream on.
	eventHandler       adsStreamEventHandler   // 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.
}

// newADSStreamImpl initializes a new adsStreamImpl instance using the given
// parameters.  It also launches goroutines responsible for managing reads and
// writes for messages of the underlying stream.
func newADSStreamImpl(opts adsStreamOpts) *adsStreamImpl {
	s := &adsStreamImpl{
		transport:          opts.transport,
		eventHandler:       opts.eventHandler,
		backoff:            opts.backoff,
		nodeProto:          opts.nodeProto,
		watchExpiryTimeout: opts.watchExpiryTimeout,

		streamCh:          make(chan clients.Stream, 1),
		runnerDoneCh:      make(chan struct{}),
		fc:                newADSFlowControl(),
		notifySender:      make(chan struct{}, 1),
		resourceTypeState: make(map[ResourceType]*resourceTypeState),
	}

	l := grpclog.Component("xds")
	s.logger = igrpclog.NewPrefixLogger(l, opts.logPrefix+fmt.Sprintf("[ads-stream %p] ", s))

	ctx, cancel := context.WithCancel(context.Background())
	s.cancel = cancel
	go s.runner(ctx)
	return s
}

// Stop blocks until the stream is closed and all spawned goroutines exit.
func (s *adsStreamImpl) Stop() {
	s.cancel()
	s.fc.stop()
	<-s.runnerDoneCh
	s.logger.Infof("Shutdown ADS stream")
}

// 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 with the
// newly subscribed resource.
func (s *adsStreamImpl) subscribe(typ ResourceType, name string) {
	if s.logger.V(2) {
		s.logger.Infof("Subscribing to resource %q of type %q", name, typ.TypeName)
	}

	s.mu.Lock()
	state, ok := s.resourceTypeState[typ]
	if !ok {
		// An entry in the type state map is created as part of the first
		// subscription request for this type.
		state = &resourceTypeState{subscribedResources: make(map[string]*xdsresource.ResourceWatchState)}
		s.resourceTypeState[typ] = state
	}

	// Create state for the newly subscribed resource. The watch timer will
	// be started when a request for this resource is actually sent out.
	state.subscribedResources[name] = &xdsresource.ResourceWatchState{State: xdsresource.ResourceWatchStateStarted}

	// Send a request for the resource type with updated subscriptions.
	s.pendingRequests = append(s.pendingRequests, request{typ: typ, resourceNames: resourceNames(state.subscribedResources)})
	s.mu.Unlock()

	select {
	case s.notifySender <- struct{}{}:
	default:
	}
}

// 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 with the updated set of resource names.
func (s *adsStreamImpl) unsubscribe(typ ResourceType, name string) {
	if s.logger.V(2) {
		s.logger.Infof("Unsubscribing to resource %q of type %q", name, typ.TypeName)
	}

	s.mu.Lock()
	state, ok := s.resourceTypeState[typ]
	if !ok {
		s.mu.Unlock()
		return
	}
	rs, ok := state.subscribedResources[name]
	if !ok {
		s.mu.Unlock()
		return
	}
	if rs.ExpiryTimer != nil {
		rs.ExpiryTimer.Stop()
	}
	delete(state.subscribedResources, name)

	// Send a request for the resource type with updated subscriptions.
	s.pendingRequests = append(s.pendingRequests, request{typ: typ, resourceNames: resourceNames(state.subscribedResources)})
	s.mu.Unlock()

	select {
	case s.notifySender <- struct{}{}:
	default:
	}
}

// runner is a long-running goroutine that handles the lifecycle of the ADS
// stream. It spawns another goroutine to handle writes of discovery request
// messages on the stream. Whenever an existing stream fails, it performs
// exponential backoff (if no messages were received on that stream) before
// creating a new stream.
func (s *adsStreamImpl) runner(ctx context.Context) {
	defer close(s.runnerDoneCh)

	go s.send(ctx)

	runStreamWithBackoff := func() error {
		stream, err := s.transport.NewStream(ctx, "/envoy.service.discovery.v3.AggregatedDiscoveryService/StreamAggregatedResources")
		if err != nil {
			s.logger.Warningf("Failed to create a new ADS streaming RPC: %v", err)
			s.streamEstablished.Store(false)
			s.onError(err, false)
			return nil
		}
		if s.logger.V(2) {
			s.logger.Infof("ADS stream created")
		}

		s.firstRequest.Store(true)
		if !s.firstStreamCreated.Load() {
			s.streamEstablished.Store(true)
			s.firstStreamCreated.Store(true)
		}

		// Ensure that the most recently created stream is pushed on the
		// channel for the `send` goroutine to consume.
		select {
		case <-s.streamCh:
		default:
		}
		s.streamCh <- stream

		// Backoff state is reset upon successful receipt of at least one
		// message from the server.
		if s.recv(stream) {
			return backoff.ErrResetBackoff
		}
		return nil
	}
	backoff.RunF(ctx, runStreamWithBackoff, s.backoff)
}

// send is a long running goroutine that handles sending discovery requests for
// two scenarios:
// - a new subscription or unsubscription request is received
// - a new stream is created after the previous one failed
func (s *adsStreamImpl) send(ctx context.Context) {
	// Stores the most recent stream instance received on streamCh.
	var stream clients.Stream
	for {
		select {
		case <-ctx.Done():
			return
		case stream = <-s.streamCh:
			if err := s.sendExisting(stream); err != nil {
				// Send failed, clear the current stream. Attempt to resend will
				// only be made after a new stream is created.
				stream = nil
				continue
			}
		case <-s.notifySender:
			// If there's no stream yet, skip the request. This request will be resent
			// when a new stream is created. If no stream is created, the watcher will
			// timeout (same as server not sending response back).
			if stream == nil {
				continue
			}

			// Resetting the pendingRequests slice to nil works for both cases:
			// - When we successfully sends the requests out on the wire.
			// - When sending fails. This can happen only when the stream fails,
			//   and in this case, we rely on the `sendExisting` to send out
			//   requests for all subscriptions when the stream is recreated.
			s.mu.Lock()
			if err := s.sendNewLocked(stream, s.pendingRequests); err != nil {
				stream = nil
			}
			s.pendingRequests = nil
			s.mu.Unlock()
		}
	}
}

// sendNewLocked attempts to send a discovery request based on a new subscription or
// unsubscription. This method also starts the watch expiry timer for resources
// that were sent in the request for the first time, i.e. their watch state is
// `watchStateStarted`.
//
// Caller needs to hold c.mu.
func (s *adsStreamImpl) sendNewLocked(stream clients.Stream, requests []request) error {
	for _, req := range requests {
		state := s.resourceTypeState[req.typ]
		if err := s.sendMessageLocked(stream, req.resourceNames, req.typ.TypeURL, state.version, state.nonce, nil); err != nil {
			return err
		}
		s.startWatchTimersLocked(req.typ, req.resourceNames)
	}
	return nil
}

// sendExisting sends out discovery requests for existing resources when
// recovering from a broken stream.
//
// The stream argument is guaranteed to be non-nil.
func (s *adsStreamImpl) sendExisting(stream clients.Stream) error {
	s.mu.Lock()
	defer s.mu.Unlock()

	// Clear any queued requests. Previously subscribed to resources will be
	// resent below.
	s.pendingRequests = nil

	for typ, state := range s.resourceTypeState {
		// Reset only the nonces map when the stream restarts.
		//
		// xDS spec says the following. See section:
		// https://www.envoyproxy.io/docs/envoy/latest/api-docs/xds_protocol#ack-nack-and-resource-type-instance-version
		//
		// Note that the version for a resource type is not a property of an
		// individual xDS stream but rather a property of the resources
		// themselves. If the stream becomes broken and the client creates a new
		// stream, the client’s initial request on the new stream should
		// indicate the most recent version seen by the client on the previous
		// stream
		state.nonce = ""

		if len(state.subscribedResources) == 0 {
			continue
		}

		names := resourceNames(state.subscribedResources)
		if err := s.sendMessageLocked(stream, names, typ.TypeURL, state.version, state.nonce, nil); err != nil {
			return err
		}
		s.startWatchTimersLocked(typ, names)
	}
	return nil
}

// sendMessageLocked sends a discovery request to the server, populating the
// different fields of the message with the given parameters. Returns a non-nil
// error if the request could not be sent.
//
// Caller needs to hold c.mu.
func (s *adsStreamImpl) sendMessageLocked(stream clients.Stream, names []string, url, version, nonce string, nackErr error) error {
	req := &v3discoverypb.DiscoveryRequest{
		ResourceNames: names,
		TypeUrl:       url,
		VersionInfo:   version,
		ResponseNonce: nonce,
	}

	// The xDS protocol only requires that we send the node proto in the first
	// discovery request on every stream. Sending the node proto in every
	// request wastes CPU resources on the client and the server.
	if s.firstRequest.Load() {
		req.Node = s.nodeProto
	}

	if nackErr != nil {
		req.ErrorDetail = &statuspb.Status{
			Code: int32(cpb.Code_INVALID_ARGUMENT), Message: nackErr.Error(),
		}
	}

	msg, err := proto.Marshal(req)
	if err != nil {
		s.logger.Warningf("Failed to marshal DiscoveryRequest: %v", err)
		return err
	}
	if err := stream.Send(msg); err != nil {
		s.logger.Warningf("Sending ADS request for type %q, resources: %v, version: %q, nonce: %q failed: %v", url, names, version, nonce, err)
		return err
	}
	s.firstRequest.Store(false)

	if s.logger.V(perRPCVerbosityLevel) {
		s.logger.Infof("ADS request sent: %v", pretty.ToJSON(req))
	} else if s.logger.V(2) {
		s.logger.Infof("ADS request sent for type %q, resources: %v, version: %q, nonce: %q", url, names, version, nonce)
	}

	return nil
}

// recv is responsible for receiving messages from the ADS stream.
//
// It performs the following actions:
//   - Waits for local flow control to be available before it receives a message
//     from the ADS stream. If an error is encountered here, it is handled by
//     the onError method which propagates the error to all watchers.
//   - Invokes the event handler's OnADSResponse method to process the message.
//   - Sends an ACK or NACK to the server based on the response.
//
// It returns a boolean indicating whether at least one message was received
// from the server.
func (s *adsStreamImpl) recv(stream clients.Stream) bool {
	msgReceived := false
	for {
		// Wait for ADS stream level flow control to be available.
		if s.fc.wait() {
			if s.logger.V(2) {
				s.logger.Infof("ADS stream stopped while waiting for flow control")
			}
			return msgReceived
		}

		resources, url, version, nonce, err := s.recvMessage(stream)
		if err != nil {
			if !msgReceived {
				s.streamEstablished.Store(false)
			}
			s.onError(err, msgReceived)
			s.logger.Warningf("ADS stream closed: %v", err)
			return msgReceived
		}
		s.streamEstablished.Store(true)
		msgReceived = true

		// Invoke the onResponse event handler to parse the incoming message and
		// decide whether to send an ACK or NACK.
		resp := response{
			resources: resources,
			typeURL:   url,
			version:   version,
		}
		var resourceNames []string
		var nackErr error
		s.fc.setPending(true)
		resourceNames, nackErr = s.eventHandler.onResponse(resp, sync.OnceFunc(func() { s.fc.setPending(false) }))
		if xdsresource.ErrType(nackErr) == xdsresource.ErrorTypeResourceTypeUnsupported {
			// A general guiding principle is that if the server sends
			// something the client didn't actually subscribe to, then the
			// client ignores it. Here, we have received a response with
			// resources of a type that we don't know about.
			//
			// Sending a NACK doesn't really seem appropriate here, since we're
			// not actually validating what the server sent and therefore don't
			// know that it's invalid.  But we shouldn't ACK either, because we
			// don't know that it is valid.
			s.logger.Warningf("%v", nackErr)
			continue
		}

		s.onRecv(stream, resourceNames, url, version, nonce, nackErr)
	}
}

func (s *adsStreamImpl) recvMessage(stream clients.Stream) (resources []*anypb.Any, url, version, nonce string, err error) {
	r, err := stream.Recv()
	if err != nil {
		return nil, "", "", "", err
	}
	var resp v3discoverypb.DiscoveryResponse
	if err := proto.Unmarshal(r, &resp); err != nil {
		s.logger.Infof("Failed to unmarshal response to DiscoveryResponse: %v", err)
		return nil, "", "", "", fmt.Errorf("unexpected message type %T", r)
	}
	if s.logger.V(perRPCVerbosityLevel) {
		s.logger.Infof("ADS response received: %v", pretty.ToJSON(&resp))
	} else if s.logger.V(2) {
		s.logger.Infof("ADS response received for type %q, version %q, nonce %q", resp.GetTypeUrl(), resp.GetVersionInfo(), resp.GetNonce())
	}
	return resp.GetResources(), resp.GetTypeUrl(), resp.GetVersionInfo(), resp.GetNonce(), nil
}

// onRecv is invoked when a response is received from the server. The arguments
// passed to this method correspond to the most recently received response.
//
// It performs the following actions:
//   - updates resource type specific state
//   - updates resource specific state for resources in the response
//   - sends an ACK or NACK to the server based on the response
func (s *adsStreamImpl) onRecv(stream clients.Stream, names []string, url, version, nonce string, nackErr error) {
	s.mu.Lock()
	defer s.mu.Unlock()

	// Lookup the resource type specific state based on the type URL.
	var typ ResourceType
	for t := range s.resourceTypeState {
		if t.TypeURL == url {
			typ = t
			break
		}
	}
	typeState, ok := s.resourceTypeState[typ]
	if !ok {
		s.logger.Warningf("ADS stream received a response for type %q, but no state exists for it", url)
		return
	}

	// Update the resource type specific state. This includes:
	//   - updating the nonce unconditionally
	//   - updating the version only if the response is to be ACKed
	previousVersion := typeState.version
	typeState.nonce = nonce
	if nackErr == nil {
		typeState.version = version
	}

	// Update the resource specific state. For all resources received as
	// part of this response that are in state `started` or `requested`,
	// this includes:
	//   - setting the watch state to watchstateReceived
	//   - stopping the expiry timer, if one exists
	for _, name := range names {
		rs, ok := typeState.subscribedResources[name]
		if !ok {
			s.logger.Warningf("ADS stream received a response for resource %q, but no state exists for it", name)
			continue
		}
		if ws := rs.State; ws == xdsresource.ResourceWatchStateStarted || ws == xdsresource.ResourceWatchStateRequested {
			rs.State = xdsresource.ResourceWatchStateReceived
			if rs.ExpiryTimer != nil {
				rs.ExpiryTimer.Stop()
				rs.ExpiryTimer = nil
			}
		}
	}

	// Send an ACK or NACK.
	subscribedResourceNames := resourceNames(typeState.subscribedResources)
	if nackErr != nil {
		s.logger.Warningf("Sending NACK for resource type: %q, version: %q, nonce: %q, reason: %v", url, version, nonce, nackErr)
		s.sendMessageLocked(stream, subscribedResourceNames, url, previousVersion, nonce, nackErr)
		return
	}

	if s.logger.V(2) {
		s.logger.Infof("Sending ACK for resource type: %q, version: %q, nonce: %q", url, version, nonce)
	}
	s.sendMessageLocked(stream, subscribedResourceNames, url, version, nonce, nil)
}

// onError is called when an error occurs on the ADS stream. It stops any
// outstanding resource timers and resets the watch state to started for any
// resources that were in the requested state. It also handles the case where
// the ADS stream was closed after receiving a response, which is not
// considered an error.
func (s *adsStreamImpl) onError(err error, msgReceived bool) {
	// For resources that been requested but not yet responded to by the
	// management server, stop the resource timers and reset the watch state to
	// watchStateStarted. This is because we don't want the expiry timer to be
	// running when we don't have a stream open to the management server.
	s.mu.Lock()
	for _, state := range s.resourceTypeState {
		for _, rs := range state.subscribedResources {
			if rs.State != xdsresource.ResourceWatchStateRequested {
				continue
			}
			if rs.ExpiryTimer != nil {
				rs.ExpiryTimer.Stop()
				rs.ExpiryTimer = nil
			}
			rs.State = xdsresource.ResourceWatchStateStarted
		}
	}
	s.mu.Unlock()

	// Note that we do not consider it an error if the ADS stream was closed
	// after having received a response on the stream. This is because there
	// are legitimate reasons why the server may need to close the stream during
	// normal operations, such as needing to rebalance load or the underlying
	// connection hitting its max connection age limit.
	// (see [gRFC A9](https://github.com/grpc/proposal/blob/master/A9-server-side-conn-mgt.md)).
	if msgReceived {
		err = xdsresource.NewError(xdsresource.ErrTypeStreamFailedAfterRecv, err.Error())
	}

	s.eventHandler.onStreamError(err)
}

// startWatchTimersLocked starts the expiry timers for the given resource names
// of the specified resource type.  For each resource name, if the resource
// watch state is in the "started" state, it transitions the state to
// "requested" and starts an expiry timer. When the timer expires, the resource
// watch state is set to "timeout" and the event handler callback is called.
//
// The caller must hold the s.mu lock.
func (s *adsStreamImpl) startWatchTimersLocked(typ ResourceType, names []string) {
	typeState := s.resourceTypeState[typ]
	for _, name := range names {
		resourceState, ok := typeState.subscribedResources[name]
		if !ok {
			continue
		}
		if resourceState.State != xdsresource.ResourceWatchStateStarted {
			continue
		}
		resourceState.State = xdsresource.ResourceWatchStateRequested

		rs := resourceState
		resourceState.ExpiryTimer = time.AfterFunc(s.watchExpiryTimeout, func() {
			s.mu.Lock()
			rs.State = xdsresource.ResourceWatchStateTimeout
			rs.ExpiryTimer = nil
			s.mu.Unlock()
			s.eventHandler.onWatchExpiry(typ, name)
		})
	}
}

func (s *adsStreamImpl) adsResourceWatchStateForTesting(rType ResourceType, resourceName string) (xdsresource.ResourceWatchState, error) {
	s.mu.Lock()
	defer s.mu.Unlock()

	state, ok := s.resourceTypeState[rType]
	if !ok {
		return xdsresource.ResourceWatchState{}, fmt.Errorf("unknown resource type: %v", rType)
	}
	resourceState, ok := state.subscribedResources[resourceName]
	if !ok {
		return xdsresource.ResourceWatchState{}, fmt.Errorf("unknown resource name: %v", resourceName)
	}
	return *resourceState, nil
}

func resourceNames(m map[string]*xdsresource.ResourceWatchState) []string {
	ret := make([]string, len(m))
	idx := 0
	for name := range m {
		ret[idx] = name
		idx++
	}
	return ret
}

// adsFlowControl implements ADS stream level flow control that enables the ADS
// stream to block the reading of the next message until the previous update is
// consumed by all watchers.
//
// The lifetime of the flow control is tied to the lifetime of the stream. When
// the stream is closed, it is the responsibility of the caller to stop the flow
// control. This ensures that any goroutine blocked on the flow control's wait
// method is unblocked.
type adsFlowControl struct {
	mu sync.Mutex
	// cond is used to signal when the most recent update has been consumed, or
	// the flow control has been stopped (in which case, waiters should be
	// unblocked as well).
	cond    *sync.Cond
	pending bool // indicates if the most recent update is pending consumption
	stopped bool // indicates if the ADS stream has been stopped
}

// newADSFlowControl returns a new adsFlowControl.
func newADSFlowControl() *adsFlowControl {
	fc := &adsFlowControl{}
	fc.cond = sync.NewCond(&fc.mu)
	return fc
}

// stop marks the flow control as stopped and signals the condition variable to
// unblock any goroutine waiting on it.
func (fc *adsFlowControl) stop() {
	fc.mu.Lock()
	defer fc.mu.Unlock()

	fc.stopped = true
	fc.cond.Broadcast()
}

// setPending changes the internal state to indicate whether there is an update
// pending consumption by all watchers. If there is no longer a pending update,
// the condition variable is signaled to allow the recv method to proceed.
func (fc *adsFlowControl) setPending(pending bool) {
	fc.mu.Lock()
	defer fc.mu.Unlock()

	if fc.stopped {
		return
	}

	fc.pending = pending
	if !pending {
		fc.cond.Broadcast()
	}
}

// wait blocks until all the watchers have consumed the most recent update.
// Returns true if the flow control was stopped while waiting, false otherwise.
func (fc *adsFlowControl) wait() bool {
	fc.mu.Lock()
	defer fc.mu.Unlock()

	for fc.pending && !fc.stopped {
		fc.cond.Wait()
	}

	return fc.stopped
}

func (s *adsStreamImpl) isStreamEstablished() bool {
	return s.streamEstablished.Load()
}