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

Code before:
				stream = nil
				continue
			}
		}
	}
}

// sendNew attempts to send a discovery request based on a new subscription or
// unsubscription. This method also starts the watch expiry timer for resources
// that were sent in the request for the first time, i.e. their watch state is
// `watchStateStarted`.
func (s *adsStreamImpl) sendNew(stream clients.Stream, typ ResourceType, names []string) error {
	s.mu.Lock()
	defer s.mu.Unlock()

	// If there's no stream yet, skip the request. This request will be resent
	// when a new stream is created. If no stream is created, the watcher will
	// timeout (same as server not sending response back).
	if stream == nil {
		return nil
	}

	state, ok := s.resourceTypeState[typ]
	if !ok {
		// State is created when the first subscription for this type is made.
		panic(fmt.Sprintf("no state exists for resource type %v", typ))
	}
	if err := s.sendMessageLocked(stream, names, typ.TypeURL, state.version, state.nonce, nil); err != nil {
		return err
	}
	s.startWatchTimersLocked(typ, names)
	return nil
}

// sendExisting sends out discovery requests for existing resources when
// recovering from a broken stream.
//
// The stream argument is guaranteed to be non-nil.
func (s *adsStreamImpl) sendExisting(stream clients.Stream) error {
	s.mu.Lock()
	defer s.mu.Unlock()

	// Clear any queued requests. Previously subscribed to resources will be
	// resent below.
	s.requestCh.Reset()

	for typ, state := range s.resourceTypeState {
		// Reset only the nonces map when the stream restarts.
		//
		// xDS spec says the following. See section:
		// https://www.envoyproxy.io/docs/envoy/latest/api-docs/xds_protocol#ack-nack-and-resource-type-instance-version

Review comment (bug):
Do we want a production panic?  Or should this be `log.Errorf()` (which fails all tests) and a `return <some error>` (or `nil` since the stream is still usable, depending on what it means to return an error?)?  The latter feels safer.

Code after:
		case <-s.notifySender:
			// If there's no stream yet, skip the request. This request will be resent
			// when a new stream is created. If no stream is created, the watcher will
			// timeout (same as server not sending response back).
			if stream == nil {
				continue
			}

			// Resetting the pendingRequests slice to nil works for both cases:
			// - When we successfully sends the requests out on the wire.
			// - When sending fails. This can happen only when the stream fails,
			//   and in this case, we rely on the `sendExisting` to send out
			//   requests for all subscriptions when the stream is recreated.
			s.mu.Lock()
			if err := s.sendNewLocked(stream, s.pendingRequests); err != nil {
				stream = nil
			}
			s.pendingRequests = nil
			s.mu.Unlock()
		}
	}
}

// sendNewLocked attempts to send a discovery request based on a new subscription or
// unsubscription. This method also starts the watch expiry timer for resources
// that were sent in the request for the first time, i.e. their watch state is
// `watchStateStarted`.
//
// Caller needs to hold c.mu.
func (s *adsStreamImpl) sendNewLocked(stream clients.Stream, requests []request) error {
	for _, req := range requests {
		state := s.resourceTypeState[req.typ]
		if err := s.sendMessageLocked(stream, req.resourceNames, req.typ.TypeURL, state.version, state.nonce, nil); err != nil {
			return err
		}
		s.startWatchTimersLocked(req.typ, req.resourceNames)
	}
	return nil
}

// sendExisting sends out discovery requests for existing resources when
// recovering from a broken stream.
//
// The stream argument is guaranteed to be non-nil.
func (s *adsStreamImpl) sendExisting(stream clients.Stream) error {
	s.mu.Lock()
	defer s.mu.Unlock()

	// Clear any queued requests. Previously subscribed to resources will be
	// resent below.
	s.pendingRequests = nil