Repository: grpc/grpc-go
File: xds/internal/clients/xdsclient/channel.go (Go)

Code before:
	// after that, and hence need not be guarded by a mutex.
	transport    clients.Transport      // Takes ownership of this transport (used to make streaming calls).
	ads          *adsStreamImpl         // An ADS stream to the management server.
	serverConfig *ServerConfig          // Configuration of the server to connect to.
	clientConfig *Config                // Complete xDS client configuration, used to decode resources.
	eventHandler xdsChannelEventHandler // Callbacks for ADS stream events.
	logger       *igrpclog.PrefixLogger // Logger to use for logging.
	closed       *syncutil.Event        // Fired when the channel is closed.
}

func (xc *xdsChannel) close() {
	if xc.closed.HasFired() {
		return
	}
	xc.closed.Fire()

	// Get the resource types that this specific ADS stream was handling
	// before stopping it.
	xc.ads.mu.Lock()
	typesHandledByStream := make([]ResourceType, 0, len(xc.ads.resourceTypeState))
	for typ := range xc.ads.resourceTypeState {
		typesHandledByStream = append(typesHandledByStream, typ)
	}
	xc.ads.mu.Unlock()

	xc.ads.Stop()

	// Schedule removeUnsubscribedCacheEntries for the types this stream was handling,
	// on all authorities that were interested in this channel.
	if _, ok := xc.eventHandler.(*channelState); ok {
		for _, typ := range typesHandledByStream {
			xc.eventHandler.requiredToRemoveUnsubscribedCacheEntries(typ)
		}
	}

	xc.transport.Close()
	xc.logger.Infof("Shutdown")
}

// subscribe adds a subscription for the given resource name of the given
// resource type on the ADS stream.
func (xc *xdsChannel) subscribe(typ ResourceType, name string) {
	if xc.closed.HasFired() {
		if xc.logger.V(2) {
			xc.logger.Infof("Attempt to subscribe to an xDS resource of type %s and name %q on a closed channel", typ.TypeName, name)
		}
		return
	}
	xc.ads.subscribe(typ, name)
}


Review comment (security):
Do we really need all this cleanup code in `close`?

An `xdsChannel` is closed *only* after all authorities have released their reference to the channel, and authority releases its reference to a channel *only* when it has no more watches. So, shouldn't the authority's cache be empty at that point?

Code after:
	// after that, and hence need not be guarded by a mutex.
	transport    clients.Transport      // Takes ownership of this transport (used to make streaming calls).
	ads          *adsStreamImpl         // An ADS stream to the management server.
	serverConfig *ServerConfig          // Configuration of the server to connect to.
	clientConfig *Config                // Complete xDS client configuration, used to decode resources.
	eventHandler xdsChannelEventHandler // Callbacks for ADS stream events.
	logger       *igrpclog.PrefixLogger // Logger to use for logging.
	closed       *syncutil.Event        // Fired when the channel is closed.
}

func (xc *xdsChannel) close() {
	if xc.closed.HasFired() {
		return
	}
	xc.closed.Fire()

	// Get the resource types that this specific ADS stream was handling
	// before stopping it.
	//
	// TODO: Revisit if we can avoid acquiring the lock of ads (another type).
	xc.ads.mu.Lock()
	typesHandledByStream := make([]ResourceType, 0, len(xc.ads.resourceTypeState))
	for typ := range xc.ads.resourceTypeState {
		typesHandledByStream = append(typesHandledByStream, typ)
	}
	xc.ads.mu.Unlock()

	xc.ads.Stop()

	// Schedule removeUnsubscribedCacheEntries for the types this stream was handling,
	// on all authorities that were interested in this channel.
	if _, ok := xc.eventHandler.(*channelState); ok {
		for _, typ := range typesHandledByStream {
			xc.eventHandler.adsResourceRemoveUnsubscribedCacheEntries(typ)
		}
	}

	xc.transport.Close()
	xc.logger.Infof("Shutdown")
}

// subscribe adds a subscription for the given resource name of the given
// resource type on the ADS stream.
func (xc *xdsChannel) subscribe(typ ResourceType, name string) {
	if xc.closed.HasFired() {
		if xc.logger.V(2) {
			xc.logger.Infof("Attempt to subscribe to an xDS resource of type %s and name %q on a closed channel", typ.TypeName, name)
		}
		return
	}
	xc.ads.subscribe(typ, name)