Function NewAsyncSharderChan

modules/frontend/pipeline/async_sharding.go:65–105

NewAsyncSharderChan creates a new AsyncResponse that shards requests to the next AsyncRoundTripper[combiner.PipelineResponse] using a limited number of goroutines.

(ctx context.Context, concurrentReqs int, reqs <-chan Request, resps Responses[combiner.PipelineResponse], next AsyncRoundTripper[combiner.PipelineResponse])

// NewAsyncSharderChan creates a new AsyncResponse that shards requests to the next AsyncRoundTripper[combiner.PipelineResponse] using a limited number of goroutines.
func NewAsyncSharderChan(ctx context.Context, concurrentReqs int, reqs <-chan Request, resps Responses[combiner.PipelineResponse], next AsyncRoundTripper[combiner.PipelineResponse]) Responses[combiner.PipelineResponse] {
	if concurrentReqs == 0 {
		panic("NewAsyncSharderChan: concurrentReqs must be greater than 0")
	}

	wg := &sync.WaitGroup{}
	asyncResp := newAsyncResponse()

	for i := 0; i < concurrentReqs; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for req := range reqs {
				if err := req.Context().Err(); err != nil {
					asyncResp.SendError(err)
					continue
				}

				resp, err := next.RoundTrip(req)
				if err != nil {
					asyncResp.SendError(err)
					continue
				}

				asyncResp.Send(ctx, resp)
			}
		}()
	}

	go func() {
		// send any responses back the caller would like to send
		if resps != nil {
			asyncResp.Send(ctx, resps)
		}

		wg.Wait()
		asyncResp.SendComplete()
	}()

	return asyncResp
}