Repository: grpc/grpc-go
File: xds/internal/balancer/cdsbalancer/cdsbalancer_test.go (Go)

Code before:
			bal := cdsBuilder.Build(bd.ClientConn, bd.BuildOptions)
			bd.ChildBalancer = bal
			cdsBalancerCh <- bal
		},
		ParseConfig: func(lbCfg json.RawMessage) (serviceconfig.LoadBalancingConfig, error) {
			return cdsBuilder.(balancer.ConfigParser).ParseConfig(lbCfg)
		},
		UpdateClientConnState: func(bd *stub.BalancerData, ccs balancer.ClientConnState) error {
			return bd.ChildBalancer.UpdateClientConnState(ccs)
		},
		Close: func(bd *stub.BalancerData) {
			bd.ChildBalancer.Close()
		},
	})
	t.Cleanup(func() { balancer.Register(cdsBuilder) })

	return cdsBalancerCh
}

// Performs the following setup required for tests:
//   - Spins up an xDS management server
//   - Creates an xDS client talking to this management server
//   - Creates a manual resolver that configures the cds LB policy as the
//     top-level policy, and pushes an initial configuration to it
//   - Creates a gRPC channel with the above manual resolver
//   - Executes OnStreamRequest callback provided by the caller
//
// Returns the following:
//   - the xDS management server
//   - the nodeID expected by the management server
//   - the grpc channel to the test backend service
//   - the manual resolver configured on the channel
//   - the xDS client used the grpc channel
func setupWithManagementServer(t *testing.T, lis net.Listener, onStreamRequest func(int64, *v3discoverypb.DiscoveryRequest) error) (*e2e.ManagementServer, string, *grpc.ClientConn, *manual.Resolver, xdsclient.XDSClient) {
	t.Helper()
	mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{
		Listener:        lis,
		OnStreamRequest: onStreamRequest,
		// Required for aggregate clusters as all resources cannot be requested
		// at once.
		AllowResourceSubset: true,
	})

	// Create bootstrap configuration pointing to the above management server.
	nodeID := uuid.New().String()
	bc := e2e.DefaultBootstrapContents(t, nodeID, mgmtServer.Address)

	config, err := bootstrap.NewConfigFromContents(bc)
	if err != nil {
		t.Fatalf("Failed to parse bootstrap contents: %s, %v", string(bc), err)
	}

Review comment (suggestion):
This doesn't make sense. This should be part of the first bullet point. Something like
- Spins up an xDS management server and passes it the onStreamRequest callback that will get invoked every time a request is received on the ADS stream.

Code after:
			bal := cdsBuilder.Build(bd.ClientConn, bd.BuildOptions)
			bd.ChildBalancer = bal
			cdsBalancerCh <- bal
		},
		ParseConfig: func(lbCfg json.RawMessage) (serviceconfig.LoadBalancingConfig, error) {
			return cdsBuilder.(balancer.ConfigParser).ParseConfig(lbCfg)
		},
		UpdateClientConnState: func(bd *stub.BalancerData, ccs balancer.ClientConnState) error {
			return bd.ChildBalancer.UpdateClientConnState(ccs)
		},
		Close: func(bd *stub.BalancerData) {
			bd.ChildBalancer.Close()
		},
	})
	t.Cleanup(func() { balancer.Register(cdsBuilder) })

	return cdsBalancerCh
}

// Performs the following setup required for tests:
//   - Spins up an xDS management server and and the provided onStreamRequest
//     function is set to be called for every incoming request on the ADS stream.
//   - Creates an xDS client talking to this management server
//   - Creates a manual resolver that configures the cds LB policy as the
//     top-level policy, and pushes an initial configuration to it
//   - Creates a gRPC channel with the above manual resolver
//
// Returns the following:
//   - the xDS management server
//   - the nodeID expected by the management server
//   - the grpc channel to the test backend service
//   - the manual resolver configured on the channel
//   - the xDS client used the grpc channel
func setupWithManagementServer(t *testing.T, lis net.Listener, onStreamRequest func(int64, *v3discoverypb.DiscoveryRequest) error) (*e2e.ManagementServer, string, *grpc.ClientConn, *manual.Resolver, xdsclient.XDSClient) {
	t.Helper()
	mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{
		Listener:        lis,
		OnStreamRequest: onStreamRequest,
		// Required for aggregate clusters as all resources cannot be requested
		// at once.
		AllowResourceSubset: true,
	})

	// Create bootstrap configuration pointing to the above management server.
	nodeID := uuid.New().String()
	bc := e2e.DefaultBootstrapContents(t, nodeID, mgmtServer.Address)

	config, err := bootstrap.NewConfigFromContents(bc)
	if err != nil {
		t.Fatalf("Failed to parse bootstrap contents: %s, %v", string(bc), err)
	}