Repository: grpc/grpc-go
File: internal/xds/clients/xdsclient/ads_stream.go (Go)

Code before:
		s.logger.Warningf("ADS request sent for type %q, resources: %v, version: %q, nonce: %q", url, names, version, nonce)
	}

	return nil
}

// recv is responsible for receiving messages from the ADS stream.
//
// It performs the following actions:
//   - Waits for local flow control to be available before sending buffered
//     requests, if any.
//   - Receives a message from the ADS stream. If an error is encountered here,
//     it is handled by the onError method which propagates the error to all
//     watchers.
//   - Invokes the event handler's OnADSResponse method to process the message.
//   - Sends an ACK or NACK to the server based on the response.
//
// It returns a boolean indicating whether at least one message was received
// from the server.
func (s *adsStreamImpl) recv(stream clients.Stream) bool {
	msgReceived := false
	for {
		// Wait for ADS stream level flow control to be available, and send out
		// a request if anything was buffered while we were waiting for local
		// processing of the previous response to complete.
		s.fc.wait()
		s.sendBuffered(stream)

		resources, url, version, nonce, err := s.recvMessage(stream)
		if err != nil {
			s.onError(err, msgReceived)
			s.logger.Warningf("ADS stream closed: %v", err)
			return msgReceived
		}
		msgReceived = true

		// Invoke the onResponse event handler to parse the incoming message and
		// decide whether to send an ACK or NACK.
		resp := response{
			resources: resources,
			typeURL:   url,
			version:   version,
		}
		var resourceNames []string
		var nackErr error
		s.fc.setPending(true)
		resourceNames, nackErr = s.eventHandler.onResponse(resp, sync.OnceFunc(func() { s.fc.setPending(false) }))
		if xdsresource.ErrType(nackErr) == xdsresource.ErrorTypeResourceTypeUnsupported {
			// A general guiding principle is that if the server sends
			// something the client didn't actually subscribe to, then the
			// client ignores it. Here, we have received a response with

Review comment (suggestion):
Do you think it helps to have `wait()` return a boolean, representing `fc.stopped`, as it did before? 
I don't think it effects correctness, but it can help avoid some warning logs when the code below sends requests on a cancelled stream.

Code after:
		s.logger.Warningf("ADS request sent for type %q, resources: %v, version: %q, nonce: %q", url, names, version, nonce)
	}

	return nil
}

// recv is responsible for receiving messages from the ADS stream.
//
// It performs the following actions:
//   - Waits for local flow control to be available before sending buffered
//     requests, if any.
//   - Receives a message from the ADS stream. If an error is encountered here,
//     it is handled by the onError method which propagates the error to all
//     watchers.
//   - Invokes the event handler's OnADSResponse method to process the message.
//   - Sends an ACK or NACK to the server based on the response.
//
// It returns a boolean indicating whether at least one message was received
// from the server.
func (s *adsStreamImpl) recv(stream clients.Stream) bool {
	msgReceived := false
	for {
		// Wait for ADS stream level flow control to be available.
		if s.fc.wait() {
			if s.logger.V(2) {
				s.logger.Infof("ADS stream stopped while waiting for flow control")
			}
			return msgReceived
		}

		// Send out a request if anything was buffered while we were waiting for
		// local processing of the previous response to complete.
		s.sendBuffered(stream)

		resources, url, version, nonce, err := s.recvMessage(stream)
		if err != nil {
			s.onError(err, msgReceived)
			s.logger.Warningf("ADS stream closed: %v", err)
			return msgReceived
		}
		msgReceived = true

		// Invoke the onResponse event handler to parse the incoming message and
		// decide whether to send an ACK or NACK.
		resp := response{
			resources: resources,
			typeURL:   url,
			version:   version,
		}
		var resourceNames []string
		var nackErr error