}

	readErrCh := make(chan error, 1)
	errCh := make(chan error, 2)

	server := newClientServerTest(t, mode, HandlerFunc(func(rw ResponseWriter, req *Request) {
		go func(body io.Reader) {
			_, err := body.Read(make([]byte, 100))
			readErrCh <- err
		}(req.Body)
		time.Sleep(500 * time.Millisecond)
	})).ts

	closeConn := make(chan bool)
	defer close(closeConn)
	go func() {

			}
			consolidatedReport.BucketStats[opts] = d
		}
	}
	return consolidatedReport
}

func (sys *NotificationSys) collectPeerMetrics(ctx context.Context, peerChannels []<-chan MetricV2, g *errgroup.Group) <-chan MetricV2 {
	ch := make(chan MetricV2)
	var wg sync.WaitGroup
	for index, err := range g.Wait() {
		if err != nil {
			if sys.peerClients[index] != nil {

package bpool

import "github.com/klauspost/reedsolomon"

// BytePoolCap implements a leaky pool of []byte in the form of a bounded channel.
type BytePoolCap struct {
	c    chan []byte
	w    int
	wcap int
}

// NewBytePoolCap creates a new BytePool bounded to the given maxSize, with new
// byte arrays sized based on width.

		},
		{`package t3; type C[T any] interface{chan<- T}; func f[T any, P C[T], Q C[[]*P]]() []T { return nil }; func _() { _ = f[int] }`,
			[]testInst{
				{`C`, []string{`T`}, `interface{chan<- T}`},
				{`C`, []string{`[]*P`}, `interface{chan<- []*P}`},
				{`f`, []string{`int`, `chan<- int`, `chan<- []*chan<- int`}, `func() []int`},
			},
		},

	MeshWatcher       mesh.Watcher
	ServiceHandler    model.ServiceHandler
	ClusterID         cluster.ID
	WatchedNamespaces string
	DomainSuffix      string
	XDSUpdater        model.XDSUpdater
	Stop              chan struct{}
	SkipRun           bool
	ConfigCluster     bool
	SystemNamespace   string
}

type FakeController struct {
	*Controller
	Endpoints *model.EndpointIndex
}

		// Register a single handler which echos the payload.
		errFatal(remote.RegisterStreamingHandler(handlerTest, StreamHandler{
			// Send 10x response.
			Handle: func(ctx context.Context, payload []byte, _ <-chan []byte, out chan<- []byte) *RemoteErr {
				for i := 0; i < responses; i++ {
					toSend := GetByteBuffer()[:0]
					toSend = append(toSend, byte(i))
					toSend = append(toSend, payload...)
					select {

		tc.nodeUpdateChannels = append(tc.nodeUpdateChannels, make(chan nodeUpdateItem, NodeUpdateChannelSize))
		tc.podUpdateChannels = append(tc.podUpdateChannels, make(chan podUpdateItem, podUpdateChannelSize))
	}

	// Functions that are responsible for taking work items out of the workqueues and putting them
	// into channels.
	go func(stopCh <-chan struct{}) {
		for {
			nodeUpdate, shutdown := tc.nodeUpdateQueue.Get()

func keyFunc(name, namespace string) string {
	if len(namespace) == 0 {
		return name
	}
	return namespace + "/" + name
}

func waitForCacheSync(name string, stop <-chan struct{}, collections ...<-chan struct{}) (r bool) {
	t := time.NewTicker(time.Second * 5)
	defer t.Stop()
	t0 := time.Now()
	defer func() {
		if r {
			log.WithLabels("name", name, "time", time.Since(t0)).Debugf("sync complete")

	sourceConfig       map[Source]TrustAnchorConfig
	mutex              sync.RWMutex
	mergedCerts        []string
	updatecb           func()
	endpointMutex      sync.RWMutex
	endpoints          []string
	endpointUpdateChan chan struct{}
	remoteCaCertPool   *x509.CertPool
	meshConfig         mesh.Watcher
}

var (
	trustBundleLog = log.RegisterScope("trustBundle", "Workload mTLS trust bundle logs")
	remoteTimeout  = 10 * time.Second
)

// If v's kind is Pointer, the pointer element type must have kind Array.
// Otherwise v's kind must be Int, Int8, Int16, Int32, Int64,
// Uint, Uint8, Uint16, Uint32, Uint64, Uintptr,
// Array, Chan, Map, Slice, or String.
func (v Value) Seq() iter.Seq[Value] {
	if canRangeFunc(v.typ()) {
		return func(yield func(Value) bool) {
			rf := MakeFunc(v.Type().In(0), func(in []Value) []Value {