2020-06-30T00:38:40.846070Z	warning	envoy config	[bazel-out/k8-opt/bin/external/envoy/source/common/config/_virtual_includes/grpc_stream_lib/common/config/grpc_stream.h:92] StreamAggregatedResources gRPC config stream closed: 13,
2020-06-30T00:39:34.522417Z	info	Subchannel Connectivity change to CONNECTING

	discovery.RegisterAggregatedDiscoveryServiceServer(grpcs, proxy)
	reflection.Register(grpcs)
	return grpcs, nil
}

const (
	TypeDebugPrefix = "istio.io/debug/"
)

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
	}

func NewAdsTest(t test.Failer, conn *grpc.ClientConn) *AdsTest {
	return NewXdsTest(t, conn, func(conn *grpc.ClientConn) (DiscoveryClient, error) {
		xds := discovery.NewAggregatedDiscoveryServiceClient(conn)
		return xds.StreamAggregatedResources(context.Background())
	})
}

func NewSdsTest(t test.Failer, conn *grpc.ClientConn) *AdsTest {
	return NewXdsTest(t, conn, func(conn *grpc.ClientConn) (DiscoveryClient, error) {

	default:
		return false
	}
}

type adsStream interface {
	Send(*discovery.DiscoveryResponse) error
	Recv() (*discovery.DiscoveryRequest, error)
	Context() context.Context
}

// StreamAggregatedResources is an implementation of XDS API used for proxying between Istiod and Envoy.
// Every time envoy makes a fresh connection to the agent, we reestablish a new connection to the upstream xds

	"istio.io/istio/pkg/test/util/retry"
)

type mockDeltaXdsServer struct{}

var deltaHandler func(stream discovery.AggregatedDiscoveryService_DeltaAggregatedResourcesServer) error

func (t *mockDeltaXdsServer) StreamAggregatedResources(discovery.AggregatedDiscoveryService_StreamAggregatedResourcesServer) error {
	return nil
}

		s.computeProxyState(con.proxy, request)
	}
	return s.pushXds(con, con.proxy.GetWatchedResource(req.TypeUrl), request)
}

// StreamAggregatedResources implements the ADS interface.
func (s *DiscoveryServer) StreamAggregatedResources(stream DiscoveryStream) error {
	return s.Stream(stream)
}

func (s *DiscoveryServer) Stream(stream DiscoveryStream) error {

	"istio.io/istio/pkg/test/util/retry"
)

type testAdscRunServer struct{}

var StreamHandler func(stream discovery.AggregatedDiscoveryService_StreamAggregatedResourcesServer) error

func (t *testAdscRunServer) StreamAggregatedResources(stream discovery.AggregatedDiscoveryService_StreamAggregatedResourcesServer) error {
	return StreamHandler(stream)
}

					grpc.WithResolvers(xdsresolver))
				if err != nil {
					t.Fatal("XDS gRPC", err)
				}
				defer conn.Close()
				s, err := discovery.NewAggregatedDiscoveryServiceClient(conn).StreamAggregatedResources(ctx)
				if err != nil {
					t.Fatal(err)
				}
				_ = s.Send(&discovery.DiscoveryRequest{})
			})
		}
	})
}

func addIstiod(sd *memory.ServiceDiscovery, xdsPort int) {

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

// Note: it is non blocking
func (a *ADSC) Run() error {
	var err error
	a.client = discovery.NewAggregatedDiscoveryServiceClient(a.conn)
	a.stream, err = a.client.StreamAggregatedResources(context.Background())
	if err != nil {
		return err
	}
	a.sendNodeMeta = true
	a.initialLoad = 0
	a.initialLds = false
	// Send the initial requests