func (b *clusterResolverBalancer) run() {
	for {
		select {
		case u := <-b.updateCh.Get():
			b.updateCh.Load()
			switch update := u.(type) {
			case *ccUpdate:
				b.handleClientConnUpdate(update)
			case *scUpdate:
				if b.child == nil {
					b.logger.Errorf("xds: received scUpdate {%+v} with no child balancer", update)
					break
				}
				b.child.UpdateSubConnState(update.subConn, update.state)
			case exitIdle:
				if b.child == nil {
					b.logger.Errorf("xds: received ExitIdle with no child balancer")
					break
				}
				if ei, ok := b.child.(balancer.ExitIdler); ok {
					ei.ExitIdle()
				}
			}
		case u := <-b.resourceWatcher.updateChannel:
			b.handleWatchUpdate(u)
		case <-b.closed.Done():
			b.resourceWatcher.stop()
			if b.child != nil {
				b.child.Close()
				b.child = nil
			}
			b.logger.Infof("Shutdown")
			b.done.Fire()
			return
		}
	}
}
run is a long-running goroutine which handles all updates from gRPC and
xdsClient. All methods which are invoked directly by gRPC or xdsClient simply
push an update onto a channel which is read and acted upon right here.
package clusterresolver
import (
	"encoding/json"
	"errors"
	"fmt"
	"google.golang.org/grpc/attributes"
	"google.golang.org/grpc/balancer"
	"google.golang.org/grpc/balancer/base"
	"google.golang.org/grpc/connectivity"
	"google.golang.org/grpc/internal/buffer"
	"google.golang.org/grpc/internal/grpclog"
	"google.golang.org/grpc/internal/grpcsync"
	"google.golang.org/grpc/internal/pretty"
	"google.golang.org/grpc/resolver"
	"google.golang.org/grpc/serviceconfig"
	"google.golang.org/grpc/xds/internal/balancer/priority"
	"google.golang.org/grpc/xds/internal/xdsclient"
)
const Name = "cluster_resolver_experimental"
var (
	errBalancerClosed = errors.New("cdsBalancer is closed")
	newChildBalancer  = func(bb balancer.Builder, cc balancer.ClientConn, o balancer.BuildOptions) balancer.Balancer {
		return bb.Build(cc, o)
	}
)
func init() {
	balancer.Register(bb{})
}
type bb struct{}
func (bb) Build(cc balancer.ClientConn, opts balancer.BuildOptions) balancer.Balancer {
	priorityBuilder := balancer.Get(priority.Name)
	if priorityBuilder == nil {
		logger.Errorf("priority balancer is needed but not registered")
		return nil
	}
	priorityConfigParser, ok := priorityBuilder.(balancer.ConfigParser)
	if !ok {
		logger.Errorf("priority balancer builder is not a config parser")
		return nil
	}
	b := &clusterResolverBalancer{
		bOpts:    opts,
		updateCh: buffer.NewUnbounded(),
		closed:   grpcsync.NewEvent(),
		done:     grpcsync.NewEvent(),
		priorityBuilder:      priorityBuilder,
		priorityConfigParser: priorityConfigParser,
	}
	b.logger = prefixLogger(b)
	b.logger.Infof("Created")
	b.resourceWatcher = newResourceResolver(b)
	b.cc = &ccWrapper{
		ClientConn:      cc,
		resourceWatcher: b.resourceWatcher,
	}
	go b.run()
	return b
}
func (bb) Name() string {
	return Name
}
func (bb) ParseConfig(c json.RawMessage) (serviceconfig.LoadBalancingConfig, error) {
	var cfg LBConfig
	if err := json.Unmarshal(c, &cfg); err != nil {
		return nil, fmt.Errorf("unable to unmarshal balancer config %s into cluster-resolver config, error: %v", string(c), err)
	}
	return &cfg, nil
}
type ccUpdate struct {
	state balancer.ClientConnState
	err   error
}
type scUpdate struct {
	subConn balancer.SubConn
	state   balancer.SubConnState
}
type exitIdle struct{}
type clusterResolverBalancer struct {
	cc              balancer.ClientConn
	bOpts           balancer.BuildOptions
	updateCh        *buffer.Unbounded  
	resourceWatcher *resourceResolver
	logger          *grpclog.PrefixLogger
	closed          *grpcsync.Event
	done            *grpcsync.Event
	priorityBuilder      balancer.Builder
	priorityConfigParser balancer.ConfigParser
	config          *LBConfig
	configRaw       *serviceconfig.ParseResult
	xdsClient       xdsclient.XDSClient     
	attrsWithClient *attributes.Attributes  
	child               balancer.Balancer
	priorities          []priorityConfig
	watchUpdateReceived bool
}
func (b *clusterResolverBalancer) handleClientConnUpdate(update *ccUpdate) {
	if err := update.err; err != nil {
		b.handleErrorFromUpdate(err, true)
		return
	}
	b.logger.Infof("Receive update from resolver, balancer config: %v", pretty.ToJSON(update.state.BalancerConfig))
	cfg, _ := update.state.BalancerConfig.(*LBConfig)
	if cfg == nil {
		b.logger.Warningf("xds: unexpected LoadBalancingConfig type: %T", update.state.BalancerConfig)
		return
	}
	b.config = cfg
	b.configRaw = update.state.ResolverState.ServiceConfig
	b.resourceWatcher.updateMechanisms(cfg.DiscoveryMechanisms)
	if !b.watchUpdateReceived {
		return
	}
	if err := b.updateChildConfig(); err != nil {
		b.logger.Warningf("failed to update child policy config: %v", err)
	}
}
func (b *clusterResolverBalancer) handleWatchUpdate(update *resourceUpdate) {
	if err := update.err; err != nil {
		b.logger.Warningf("Watch error from xds-client %p: %v", b.xdsClient, err)
		b.handleErrorFromUpdate(err, false)
		return
	}
	b.logger.Infof("resource update: %+v", pretty.ToJSON(update.priorities))
	b.watchUpdateReceived = true
	b.priorities = update.priorities
	if err := b.updateChildConfig(); err != nil {
		b.logger.Warningf("failed to update child policy's balancer config: %v", err)
	}
}
func (b *clusterResolverBalancer) updateChildConfig() error {
	if b.child == nil {
		b.child = newChildBalancer(b.priorityBuilder, b.cc, b.bOpts)
	}
	childCfgBytes, addrs, err := buildPriorityConfigJSON(b.priorities, b.config.XDSLBPolicy)
	if err != nil {
		return fmt.Errorf("failed to build priority balancer config: %v", err)
	}
	childCfg, err := b.priorityConfigParser.ParseConfig(childCfgBytes)
	if err != nil {
		return fmt.Errorf("failed to parse generated priority balancer config, this should never happen because the config is generated: %v", err)
	}
	b.logger.Infof("build balancer config: %v", pretty.ToJSON(childCfg))
	return b.child.UpdateClientConnState(balancer.ClientConnState{
		ResolverState: resolver.State{
			Addresses:     addrs,
			ServiceConfig: b.configRaw,
			Attributes:    b.attrsWithClient,
		},
		BalancerConfig: childCfg,
	})
}
func (b *clusterResolverBalancer) handleErrorFromUpdate(err error, fromParent bool) {
	b.logger.Warningf("Received error: %v", err)
	if fromParent && xdsclient.ErrType(err) == xdsclient.ErrorTypeResourceNotFound {
		b.resourceWatcher.stop()
	}
	if b.child != nil {
		b.child.ResolverError(err)
	} else {
		b.cc.UpdateState(balancer.State{
			ConnectivityState: connectivity.TransientFailure,
			Picker:            base.NewErrPicker(err),
		})
	}
}