1func (s *DiscoveryServer) StreamAggregatedResources(stream discovery.AggregatedDiscoveryService_StreamAggregatedResourcesServer) error {
 2	return s.Stream(stream)
 3}
 4
 5func (s *DiscoveryServer) Stream(stream DiscoveryStream) error {
 6	...
 7  // 建立新的连接
 8	con := newConnection(peerAddr, stream)
 9	...
10  // 创建一个接受请求的channel
11	reqChannel := make(chan *discovery.DiscoveryRequest, 1)
12  // 处理客户端的请求
13	go s.receive(con, reqChannel, &receiveError)
14
15	// Wait for the proxy to be fully initialized before we start serving traffic. Because
16	// initialization doesn't have dependencies that will block, there is no need to add any timeout
17	// here. Prior to this explicit wait, we were implicitly waiting by receive() not sending to
18	// reqChannel and the connection not being enqueued for pushes to pushChannel until the
19	// initialization is complete.
20	<-con.initialized
21
22	for {
23		// Block until either a request is received or a push is triggered.
24		// We need 2 go routines because 'read' blocks in Recv().
25		//
26		// To avoid 2 routines, we tried to have Recv() in StreamAggregateResource - and the push
27		// on different short-lived go routines started when the push is happening. This would cut in 1/2
28		// the number of long-running go routines, since push is throttled. The main problem is with
29		// closing - the current gRPC library didn't allow closing the stream.
30		select {
31		case req, ok := <-reqChannel:
32			if !ok {
33				// Remote side closed connection or error processing the request.
34				return receiveError
35			}
36			// processRequest is calling pushXXX, accessing common structs with pushConnection.
37			// Adding sync is the second issue to be resolved if we want to save 1/2 of the threads.
38			err := s.processRequest(req, con)
39			if err != nil {
40				return err
41			}
42
43    // 获取从 Server 端推送的 PushRequest 相关Event
44		case pushEv := <-con.pushChannel:
45			err := s.pushConnection(con, pushEv)
46			pushEv.done()
47			if err != nil {
48				return err
49			}
50		case <-con.stop:
51			return nil
52		}
53	}
54}