Method runStream

orca/producer.go:203–225

runStream runs a single stream on the subchannel and returns the resulting error, if any, and whether or not the run loop should reset the backoff timer to zero or advance it.

(ctx context.Context, interval time.Duration)

// error, if any, and whether or not the run loop should reset the backoff
// timer to zero or advance it.
func (p *producer) runStream(ctx context.Context, interval time.Duration) (resetBackoff bool, err error) {
	streamCtx, cancel := context.WithCancel(ctx)
	defer cancel()
	stream, err := p.client.StreamCoreMetrics(streamCtx, &v3orcaservicepb.OrcaLoadReportRequest{
		ReportInterval: durationpb.New(interval),
	})
	if err != nil {
		return false, err
	}

	for {
		report, err := stream.Recv()
		if err != nil {
			return resetBackoff, err
		}
		resetBackoff = true
		p.mu.Lock()
		for l := range p.listeners {
			l.OnLoadReport(report)
		}
		p.mu.Unlock()
	}
}