defer log.Debugf("disconnected from delta XDS server: %s", p.istiodAddress)

	con.upstreamDeltas = deltaUpstream

	// handle responses from istiod
	go func() {
		for {
			resp, err := con.upstreamDeltas.Recv()
			if err != nil {
				upstreamErr(con, err)
				return
			}
			select {
			case con.deltaResponsesChan <- resp:
			case <-con.stopChan:
			}
		}
	}()

	go p.handleUpstreamDeltaRequest(con)

	defer func() {
		close(con.deltaReqChan)
		close(con.ErrorCh())
		// Close the initialized channel, if its not already closed, to prevent blocking the stream
		select {
		case <-con.InitializedCh():
		default:
			close(con.InitializedCh())
		}
	}()
	firstRequest := true
	for {
		req, err := con.deltaStream.Recv()
		if err != nil {
			if istiogrpc.IsExpectedGRPCError(err) {

		return s.pushXds(con,
			&model.WatchedResource{TypeUrl: req.TypeUrl, ResourceNames: req.ResourceNames},
			&model.PushRequest{Full: true, Push: con.proxy.LastPushContext})
	}

	if s.StatusReporter != nil {
		s.StatusReporter.RegisterEvent(con.ID(), req.TypeUrl, req.ResponseNonce)
	}

	shouldRespond, delta := xds.ShouldRespond(con.proxy, con.ID(), req)
	if !shouldRespond {
		return nil
	}

				ClusterSent:          con.NonceSent(v3.ClusterType),
				ClusterAcked:         con.NonceAcked(v3.ClusterType),
				ListenerSent:         con.NonceSent(v3.ListenerType),
				ListenerAcked:        con.NonceAcked(v3.ListenerType),
				RouteSent:            con.NonceSent(v3.RouteType),
				RouteAcked:           con.NonceAcked(v3.RouteType),
				EndpointSent:         con.NonceSent(v3.EndpointType),

func Receive(ctx ConnectionContext) {
	con := ctx.XdsConnection()
	defer func() {
		close(con.errorChan)
		close(con.reqChan)
		// Close the initialized channel, if its not already closed, to prevent blocking the stream.
		select {
		case <-con.initialized:
		default:
			close(con.initialized)
		}
	}()

	firstRequest := true
	for {
		req, err := con.stream.Recv()
		if err != nil {

// isSidecar ad-hoc method to see if connection represents a sidecar
func isProxy(con *Connection) bool {
	return con != nil &&
		con.proxy != nil &&
		con.proxy.Metadata != nil &&
		con.proxy.Metadata.ProxyConfig != nil
}

func isZtunnel(con *Connection) bool {
	return con != nil &&
		con.proxy != nil &&
		con.proxy.Metadata != nil &&
		con.proxy.Type == model.Ztunnel
}

	assert.Nil(t, con.UserDefinedMetrics)

	con = indexCon[cName01]
	assert.EqualValues(t, testTime(creationTime, seedPod0Container1).Unix(), con.StartTime.Time.Unix())
	checkCPUStats(t, "Pod0Container1", seedPod0Container1, con.CPU)
	checkMemoryStats(t, "Pod0Container1", seedPod0Container1, infos["/pod0-c1"], con.Memory)
	assert.Nil(t, con.Rootfs)
	assert.Nil(t, con.Logs)
	assert.Nil(t, con.Accelerators)

	con.upstream = upstream

	// Handle upstream xds recv
	go func() {
		for {
			// from istiod
			resp, err := con.upstream.Recv()
			if err != nil {
				upstreamErr(con, err)
				return
			}
			select {
			case con.responsesChan <- resp:
			case <-con.stopChan:
			}
		}
	}()

	go p.handleUpstreamRequest(con)
	go p.handleUpstreamResponse(con)

	}
	return nil, err
}

func (s *DiscoveryServer) authorize(con *Connection, identities []string) error {
	if con == nil || con.proxy == nil {
		return nil
	}

	if features.EnableXDSIdentityCheck && identities != nil {
		// TODO: allow locking down, rejecting unauthenticated requests.
		id, err := checkConnectionIdentity(con.proxy, identities)
		if err != nil {

Tú declaras la "forma" de los datos mediante clases con atributos.

Cada atributo tiene un tipo.

Luego creas un instance de esa clase con algunos valores y Pydantic validará los valores, los convertirá al tipo apropiado (si ese es el caso) y te dará un objeto con todos los datos.

Y obtienes todo el soporte del editor con el objeto resultante.

Tomado de la documentación oficial de Pydantic: