t.Helper()
	adsClient := discovery.NewAggregatedDiscoveryServiceClient(conn)
	downstream, err := adsClient.DeltaAggregatedResources(ctx)
	if err != nil {
		t.Fatal(err)
	}
	return downstream
}

func sendDeltaDownstreamWithNode(t *testing.T, downstream discovery.AggregatedDiscoveryService_DeltaAggregatedResourcesClient, meta model.NodeMetadata) {
	t.Helper()
	node := &core.Node{

		conn := setupDownstreamConnection(t, proxy)
		downstream := stream(t, conn)
		sendDownstreamWithNode(t, downstream, model.NodeMetadata{
			Namespace:   "default",
			InstanceIPs: []string{"1.1.1.1"},
		})

		downstream.CloseSend()
		waitDisconnect(proxy)

		downstream = stream(t, conn)
		sendDownstreamWithNode(t, downstream, model.NodeMetadata{
			Namespace:   "default",

)

func (p *tapProxy) StreamAggregatedResources(downstream DiscoveryStream) error {
	timeout := time.Second * 15
	req, err := downstream.Recv()
	if err != nil {
		log.Errorf("failed to recv: %v", err)
		return err
	}
	if strings.HasPrefix(req.TypeUrl, TypeDebugPrefix) {
		if resp, err := p.xdsProxy.tapRequest(req, timeout); err == nil {
			err := downstream.Send(resp)
			if err != nil {

// as the new connection may not go to the same istiod. Vice versa case also applies.
func (p *XdsProxy) StreamAggregatedResources(downstream DiscoveryStream) error {
	proxyLog.Debugf("accepted XDS connection from Envoy, forwarding to upstream XDS server")
	return p.handleStream(downstream)
}

func (p *XdsProxy) handleStream(downstream adsStream) error {
	con := &ProxyConnection{
		conID:           connectionNumber.Inc(),

	go p.handleUpstreamDeltaResponse(con)

	for {
		select {
		case err := <-con.upstreamError:
			return err
		case err := <-con.downstreamError:
			// On downstream error, we will return. This propagates the error to downstream envoy which will trigger reconnect
			return err
		case <-con.stopChan:
			log.Debugf("upstream stopped")
			return nil
		}
	}
}

    is enabled by default. This means SNI will be set automatically based on the downstream HTTP host/authority header if `DestinationRule` does not explicitly set the same.

                             const QuantizationSpecs& quantization_specs,
                             const DebuggerConfig& debugger_config) {
  // Convert NCHW tensors to NHWC at along with extra optimizations as
  // downstream passes perform better optimizations when dealing with NHWC
  // formatted tensors.
  AddProcessNchwTensorPasses(pm);

  pm.addPass(CreateLiftQuantizableSpotsAsFunctionsPass(quantization_specs));

// WebSocket data frame. The primary structure for this tunneling is
// the TunnelingConnection. A lot of the other code in streamtunnel.go
// is for properly upgrading both the upstream SPDY connection and the
// downstream WebSocket connection before streaming begins.

import okio.Timeout

/**
 * Replicates a single upstream source into multiple downstream sources. Each downstream source
 * returns the same bytes as the upstream source. Downstream sources may read data either as it
 * is returned by upstream, or after the upstream source has been exhausted.
 *
 * As bytes are returned from upstream they are written to a local file. Downstream sources read
 * from this file as necessary.
 *

  `ECDHE-ECDSA-AES256-GCM-SHA384`, `ECDHE-RSA-AES256-GCM-SHA384`, and ECDH
  curves to `P-256`. These restrictions apply on the following data paths:

  * mTLS communication between Envoy proxies;
  * regular TLS on the downstream and the upstream of Envoy proxies (e.g. gateway);
  * Google gRPC side requests from Envoy proxies (e.g. Stackdriver extensions);
  * Istiod xDS server;
  * Istiod injection and validation webhook servers.