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

Code before:
//
// 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
			// resources of a type that we don't know about.
			//
			// Sending a NACK doesn't really seem appropriate here, since we're
			// not actually validating what the server sent and therefore don't
			// know that it's invalid.  But we shouldn't ACK either, because we
			// don't know that it is valid.
			s.logger.Warningf("%v", nackErr)

Review comment (bug):
Not related to changes in this PR, `sendBuffered` returns an error which is being ignored here. Does the returned error indicate that the stream is broken? If yes, should we return early? This can be done in a separate PR to keep this PR focused on the bug fix.

Code after:
//
// 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
		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
			// resources of a type that we don't know about.