xdsclient: modify how the resource watch state is retrieved for testing (#8499)

Author: easwars

xds/internal/clients/xdsclient/authority.go

@@ -20,6 +20,7 @@ package xdsclient
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"sync"
 	"sync/atomic"
@@ -955,3 +956,17 @@ func serviceStatusToProto(serviceStatus xdsresource.ServiceStatus) v3adminpb.Cli
 		return v3adminpb.ClientResourceStatus_UNKNOWN
 	}
 }
+
+func (a *authority) resourceWatchStateForTesting(rType ResourceType, resourceName string) (state xdsresource.ResourceWatchState, err error) {
+	done := make(chan struct{})
+	a.xdsClientSerializer.ScheduleOr(func(context.Context) {
+		state, err = a.activeXDSChannel.channel.ads.adsResourceWatchStateForTesting(rType, resourceName)
+		close(done)
+	}, func() {
+		err = errors.New("failed to retrieve resource watch state because the xDS client is closed")
+		close(done)
+	})
+	<-done
+
+	return state, err
+}

xds/internal/clients/xdsclient/xdsclient.go

@@ -453,13 +453,11 @@ func (cs *channelState) adsResourceRemoveUnsubscribedCacheEntries(rType Resource
 }
 
 func resourceWatchStateForTesting(c *XDSClient, rType ResourceType, resourceName string) (xdsresource.ResourceWatchState, error) {
-	c.channelsMu.Lock()
-	defer c.channelsMu.Unlock()
-
-	for _, state := range c.xdsActiveChannels {
-		if st, err := state.channel.ads.adsResourceWatchStateForTesting(rType, resourceName); err == nil {
-			return st, nil
-		}
+	n := xdsresource.ParseName(resourceName)
+	a := c.getAuthorityForResource(n)
+	if a == nil {
+		return xdsresource.ResourceWatchState{}, fmt.Errorf("unable to find authority for resource name %q", resourceName)
 	}
-	return xdsresource.ResourceWatchState{}, fmt.Errorf("unable to find watch state for resource type %q and name %q", rType.TypeName, resourceName)
+	return a.resourceWatchStateForTesting(rType, resourceName)
+
 }