}
			return
		}
		select {
		case a.responses <- resp:
		case <-a.context.Done():
			return
		}
	}
}

// DrainResponses reads all responses, but does nothing to them
func (a *DeltaAdsTest) DrainResponses() {
	a.t.Helper()
	for {
		select {
		case <-a.context.Done():
			return
		case r := <-a.responses:
			log.Infof("drained response %v", r.TypeUrl)
		}
	}
}

			}
			return
		}
		select {
		case a.responses <- resp:
		case <-a.context.Done():
			return
		}
	}
}

// DrainResponses reads all responses, but does nothing to them
func (a *AdsTest) DrainResponses() {
	for {
		select {
		case <-a.context.Done():
			return
		case r := <-a.responses:
			log.Infof("drained response %v", r.TypeUrl)
		}
	}
}

			err = fmt.Errorf("unexpected number of responses: expected %d, received %d",
				opts.Count, len(responses))
		}

		// Convert to a CallResult.
		result := echo.CallResult{
			From:      from,
			Opts:      opts,
			Responses: responses,
		}

		// Return the results from the validator.
		err = opts.Check(result, err)
		if err != nil {
			err = fmt.Errorf("call failed from %s to %s (using %s): %v",

	)

	// TODO: Add type url as type for requeasts and responses if needed.

	// XdsProxyRequests records total number of downstream requests.
	XdsProxyRequests = monitoring.NewSum(
		"xds_proxy_requests",
		"The total number of Xds Proxy Requests",
	)

	// XdsProxyResponses records total number of upstream responses.
	XdsProxyResponses = monitoring.NewSum(
		"xds_proxy_responses",

			Name: port,
		},
		Scheme: scheme.TLS,
		TLS: echo.TLS{
			Alpn: []string{"istio"},
		},
		Check: check.NoError(),
	})
	if result.Responses.Len() != 1 {
		t.Fatalf("dump cert failed, no responses")
	}
	var certs []string
	for _, rr := range result.Responses[0].Body() {
		var s string
		if err := json.Unmarshal([]byte(rr), &s); err != nil {
			t.Fatalf("failed to unmarshal: %v", err)
		}

		err := json.Unmarshal(resp, istiodDump)
		if err != nil {
			continue
		}
		c.istiod = istiodDump
		break
	}
	if c.istiod == nil {
		return nil, fmt.Errorf("unable to find config dump in Istiod responses")
	}
	envoyDump := &configdump.Wrapper{}
	err := json.Unmarshal(envoyResponse, envoyDump)
	if err != nil {
		return nil, err
	}
	c.envoy = envoyDump
	c.w = w
	c.context = 7

		"If true, Pilot will cache XDS responses.").Get()

	// EnableCDSCaching determines if CDS caching is enabled. This is explicitly split out of ENABLE_XDS_CACHE,
	// so that in case there are issues with the CDS cache we can just disable the CDS cache.
	EnableCDSCaching = env.Register("PILOT_ENABLE_CDS_CACHE", true,
		"If true, Pilot will cache CDS responses. Note: this depends on PILOT_ENABLE_XDS_CACHE.").Get()

		}

		return &proto.ForwardEchoResponse{
			Output: responses,
		}, nil
	case <-ctx.Done():
		responsesMu.Lock()
		defer responsesMu.Unlock()

		var c int
		var tt time.Duration
		for id, res := range responses {
			if res != "" && responseTimes[id] != 0 {
				c++
				tt += responseTimes[id]
			}
		}
		var avgTime time.Duration

	"istio.io/istio/pilot/pkg/model"
	xdsresource "istio.io/istio/pilot/pkg/xds/v3"
	"istio.io/istio/pkg/log"
)

// XdsStatusWriter enables printing of sync status using multiple xdsapi.DiscoveryResponse Istiod responses
type XdsStatusWriter struct {
	Writer                 io.Writer
	Namespace              string
	InternalDebugAllIstiod bool
}

type xdsWriterStatus struct {
	proxyID               string

		Server: s,
	}
}

// Generate XDS responses about internal events:
// - connection status
// - NACKs
// We can also expose ACKS.
func (sg *StatusGen) Generate(proxy *model.Proxy, w *model.WatchedResource, req *model.PushRequest) (model.Resources, model.XdsLogDetails, error) {
	return sg.handleInternalRequest(proxy, w, req)
}

// Generate delta XDS responses about internal events:
// - connection status
// - NACKs