func (c *clientImpl) WatchEndpoints(clusterName string, cb func(EndpointsUpdate, error)) (cancel func()) {
	wi := &watchInfo{
		c:           c,
		rType:       EndpointsResource,
		target:      clusterName,
		edsCallback: cb,
	}
	wi.expiryTimer = time.AfterFunc(c.watchExpiryTimeout, func() {
		wi.timeout()
	})
	return c.watch(wi)
}
WatchEndpoints uses EDS to discover endpoints in the provided clusterName.

WatchEndpoints can be called multiple times, with same or different
clusterNames. Each call will start an independent watcher for the resource.

Note that during race (e.g. an xDS response is received while the user is
calling cancel()), there's a small window where the callback can be called
after the watcher is canceled. The caller needs to handle this case.
package xdsclient
import (
	"fmt"
	"sync"
	"time"
	"google.golang.org/grpc/internal/pretty"
)
type watchInfoState int
const (
	watchInfoStateStarted watchInfoState = iota
	watchInfoStateRespReceived
	watchInfoStateTimeout
	watchInfoStateCanceled
)
type watchInfo struct {
	c      *clientImpl
	rType  ResourceType
	target string
	ldsCallback func(ListenerUpdate, error)
	rdsCallback func(RouteConfigUpdate, error)
	cdsCallback func(ClusterUpdate, error)
	edsCallback func(EndpointsUpdate, error)
	expiryTimer *time.Timer
	mu    sync.Mutex
	state watchInfoState
}
func (wi *watchInfo) newUpdate(update interface{}) {
	wi.mu.Lock()
	defer wi.mu.Unlock()
	if wi.state == watchInfoStateCanceled {
		return
	}
	wi.state = watchInfoStateRespReceived
	wi.expiryTimer.Stop()
	wi.c.scheduleCallback(wi, update, nil)
}
func (wi *watchInfo) newError(err error) {
	wi.mu.Lock()
	defer wi.mu.Unlock()
	if wi.state == watchInfoStateCanceled {
		return
	}
	wi.state = watchInfoStateRespReceived
	wi.expiryTimer.Stop()
	wi.sendErrorLocked(err)
}
func (wi *watchInfo) resourceNotFound() {
	wi.mu.Lock()
	defer wi.mu.Unlock()
	if wi.state == watchInfoStateCanceled {
		return
	}
	wi.state = watchInfoStateRespReceived
	wi.expiryTimer.Stop()
	wi.sendErrorLocked(NewErrorf(ErrorTypeResourceNotFound, "xds: %v target %s not found in received response", wi.rType, wi.target))
}
func (wi *watchInfo) timeout() {
	wi.mu.Lock()
	defer wi.mu.Unlock()
	if wi.state == watchInfoStateCanceled || wi.state == watchInfoStateRespReceived {
		return
	}
	wi.state = watchInfoStateTimeout
	wi.sendErrorLocked(fmt.Errorf("xds: %v target %s not found, watcher timeout", wi.rType, wi.target))
}
func (wi *watchInfo) sendErrorLocked(err error) {
	var (
		u interface{}
	)
	switch wi.rType {
	case ListenerResource:
		u = ListenerUpdate{}
	case RouteConfigResource:
		u = RouteConfigUpdate{}
	case ClusterResource:
		u = ClusterUpdate{}
	case EndpointsResource:
		u = EndpointsUpdate{}
	}
	wi.c.scheduleCallback(wi, u, err)
}
func (wi *watchInfo) cancel() {
	wi.mu.Lock()
	defer wi.mu.Unlock()
	if wi.state == watchInfoStateCanceled {
		return
	}
	wi.expiryTimer.Stop()
	wi.state = watchInfoStateCanceled
}
func (c *clientImpl) watch(wi *watchInfo) (cancel func()) {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.logger.Debugf("new watch for type %v, resource name %v", wi.rType, wi.target)
	var (
		watchers map[string]map[*watchInfo]bool
		mds      map[string]UpdateMetadata
	)
	switch wi.rType {
	case ListenerResource:
		watchers = c.ldsWatchers
		mds = c.ldsMD
	case RouteConfigResource:
		watchers = c.rdsWatchers
		mds = c.rdsMD
	case ClusterResource:
		watchers = c.cdsWatchers
		mds = c.cdsMD
	case EndpointsResource:
		watchers = c.edsWatchers
		mds = c.edsMD
	default:
		c.logger.Errorf("unknown watch type: %v", wi.rType)
		return nil
	}
	resourceName := wi.target
	s, ok := watchers[wi.target]
	if !ok {
		c.logger.Debugf("first watch for type %v, resource name %v, will send a new xDS request", wi.rType, wi.target)
		s = make(map[*watchInfo]bool)
		watchers[resourceName] = s
		mds[resourceName] = UpdateMetadata{Status: ServiceStatusRequested}
		c.apiClient.AddWatch(wi.rType, resourceName)
	}
	s[wi] = true
	switch wi.rType {
	case ListenerResource:
		if v, ok := c.ldsCache[resourceName]; ok {
			c.logger.Debugf("LDS resource with name %v found in cache: %+v", wi.target, pretty.ToJSON(v))
			wi.newUpdate(v)
		}
	case RouteConfigResource:
		if v, ok := c.rdsCache[resourceName]; ok {
			c.logger.Debugf("RDS resource with name %v found in cache: %+v", wi.target, pretty.ToJSON(v))
			wi.newUpdate(v)
		}
	case ClusterResource:
		if v, ok := c.cdsCache[resourceName]; ok {
			c.logger.Debugf("CDS resource with name %v found in cache: %+v", wi.target, pretty.ToJSON(v))
			wi.newUpdate(v)
		}
	case EndpointsResource:
		if v, ok := c.edsCache[resourceName]; ok {
			c.logger.Debugf("EDS resource with name %v found in cache: %+v", wi.target, pretty.ToJSON(v))
			wi.newUpdate(v)
		}
	}
	return func() {
		c.logger.Debugf("watch for type %v, resource name %v canceled", wi.rType, wi.target)
		wi.cancel()
		c.mu.Lock()
		defer c.mu.Unlock()
		if s := watchers[resourceName]; s != nil {
			delete(s, wi)
			if len(s) == 0 {
				c.logger.Debugf("last watch for type %v, resource name %v canceled, will send a new xDS request", wi.rType, wi.target)
				delete(watchers, resourceName)
				delete(mds, resourceName)
				c.apiClient.RemoveWatch(wi.rType, resourceName)
				switch wi.rType {
				case ListenerResource:
					delete(c.ldsCache, resourceName)
				case RouteConfigResource:
					delete(c.rdsCache, resourceName)
				case ClusterResource:
					delete(c.cdsCache, resourceName)
				case EndpointsResource:
					delete(c.edsCache, resourceName)
				}
			}
		}
	}
}
func (c *clientImpl) WatchListener(serviceName string, cb func(ListenerUpdate, error)) (cancel func()) {
	wi := &watchInfo{
		c:           c,
		rType:       ListenerResource,
		target:      serviceName,
		ldsCallback: cb,
	}
	wi.expiryTimer = time.AfterFunc(c.watchExpiryTimeout, func() {
		wi.timeout()
	})
	return c.watch(wi)
}
func (c *clientImpl) WatchRouteConfig(routeName string, cb func(RouteConfigUpdate, error)) (cancel func()) {
	wi := &watchInfo{
		c:           c,
		rType:       RouteConfigResource,
		target:      routeName,
		rdsCallback: cb,
	}
	wi.expiryTimer = time.AfterFunc(c.watchExpiryTimeout, func() {
		wi.timeout()
	})
	return c.watch(wi)
}
func (c *clientImpl) WatchCluster(clusterName string, cb func(ClusterUpdate, error)) (cancel func()) {
	wi := &watchInfo{
		c:           c,
		rType:       ClusterResource,
		target:      clusterName,
		cdsCallback: cb,
	}
	wi.expiryTimer = time.AfterFunc(c.watchExpiryTimeout, func() {
		wi.timeout()
	})
	return c.watch(wi)
}