xdsclient: complete refactor and fallback support

internal/internal.go

@@ -149,6 +149,20 @@ var (
 	// other features, including the CSDS service.
 	NewXDSResolverWithConfigForTesting any // func([]byte) (resolver.Builder, error)
 
+	// NewXDSResolverWithClientForTesting creates a new xDS resolver builder
+	// using the provided xDS client instead of creating a new one using the
+	// bootstrap configuration specified by the supported environment variables.
+	// The resolver.Builder is meant to be used in conjunction with the
+	// grpc.WithResolvers DialOption. The resolver.Builder does not take
+	// ownership of the provided xDS client and it is the responsibility of the
+	// caller to close the client when no longer required.
+	//
+	// Testing Only
+	//
+	// This function should ONLY be used for testing and may not work with some
+	// other features, including the CSDS service.
+	NewXDSResolverWithClientForTesting any // func(xdsclient.XDSClient) (resolver.Builder, error)
+
 	// RegisterRLSClusterSpecifierPluginForTesting registers the RLS Cluster
 	// Specifier Plugin for testing purposes, regardless of the XDSRLS environment
 	// variable.

internal/testutils/restartable_listener.go

@@ -81,6 +81,7 @@ func (l *RestartableListener) Addr() net.Addr {
 // Stop closes existing connections on the listener and prevents new connections
 // from being accepted.
 func (l *RestartableListener) Stop() {
+	logger.Infof("Stopping restartable listener %q", l.Addr())
 	l.mu.Lock()
 	l.stopped = true
 	for _, conn := range l.conns {
@@ -92,6 +93,7 @@ func (l *RestartableListener) Stop() {
 
 // Restart gets a previously stopped listener to start accepting connections.
 func (l *RestartableListener) Restart() {
+	logger.Infof("Restarting listener %q", l.Addr())
 	l.mu.Lock()
 	l.stopped = false
 	l.mu.Unlock()

internal/testutils/wrappers.go

@@ -53,7 +53,7 @@ func (l *ListenerWrapper) Accept() (net.Conn, error) {
 	}
 	closeCh := NewChannel()
 	conn := &ConnWrapper{Conn: c, CloseCh: closeCh}
-	l.NewConnCh.Send(conn)
+	l.NewConnCh.Replace(conn)
 	return conn, nil
 }
 

xds/internal/resolver/xds_resolver.go

@@ -44,20 +44,34 @@ import (
 // xdsresolver.Scheme
 const Scheme = "xds"
 
-// newBuilderForTesting creates a new xds resolver builder using a specific xds
-// bootstrap config, so tests can use multiple xds clients in different
-// ClientConns at the same time.
-func newBuilderForTesting(config []byte) (resolver.Builder, error) {
+// newBuilderWithConfigForTesting creates a new xds resolver builder using a
+// specific xds bootstrap config, so tests can use multiple xds clients in
+// different ClientConns at the same time.
+func newBuilderWithConfigForTesting(config []byte) (resolver.Builder, error) {
 	return &xdsResolverBuilder{
 		newXDSClient: func(name string) (xdsclient.XDSClient, func(), error) {
 			return xdsclient.NewForTesting(xdsclient.OptionsForTesting{Name: name, Contents: config})
 		},
 	}, nil
 }
 
+// newBuilderWithClientForTesting creates a new xds resolver builder using the
+// specific xDS client, so that tests have complete control over the exact
+// specific xDS client being used.
+func newBuilderWithClientForTesting(client xdsclient.XDSClient) (resolver.Builder, error) {
+	return &xdsResolverBuilder{
+		newXDSClient: func(string) (xdsclient.XDSClient, func(), error) {
+			// Returning an empty close func here means that the responsibility
+			// of closing the client lies with the caller.
+			return client, func() {}, nil
+		},
+	}, nil
+}
+
 func init() {
 	resolver.Register(&xdsResolverBuilder{})
-	internal.NewXDSResolverWithConfigForTesting = newBuilderForTesting
+	internal.NewXDSResolverWithConfigForTesting = newBuilderWithConfigForTesting
+	internal.NewXDSResolverWithClientForTesting = newBuilderWithClientForTesting
 
 	rinternal.NewWRR = wrr.NewRandom
 	rinternal.NewXDSClient = xdsclient.New