Method RoundTrip

modules/frontend/pipeline/collector_grpc.go:33–86

RoundTrip implements the http.RoundTripper interface

(req *http.Request)

// RoundTrip implements the http.RoundTripper interface
func (c GRPCCollector[T]) RoundTrip(req *http.Request) error {
	ctx := req.Context()
	ctx, cancel := context.WithCancel(ctx) // create a new context with a cancel function
	defer cancel()

	ctx, span := tracer.Start(ctx, "GRPCCollector.RoundTrip")
	defer span.End()

	req = req.WithContext(ctx)
	resps, err := c.next.RoundTrip(NewHTTPRequest(req))
	if err != nil {
		return grpcError(err)
	}
	span.AddEvent("next.RoundTrip done")

	lastUpdate := time.Now()
	// sendDiffCb should return an error if the context is cancelled,
	// callback's error is used to exit early from the loop and return the error to the caller
	sendDiffCb := func() error {
		// check if we should send an update
		if time.Since(lastUpdate) > 500*time.Millisecond {
			lastUpdate = time.Now()
			// check and return the context errors, like ctx cancelled, etc
			if req.Context().Err() != nil {
				return req.Context().Err()
			}

			// send a diff only during streaming
			resp, err := c.combiner.GRPCDiff()
			if err != nil {
				return err
			}

			return c.sendSegmented(req, resp, c.maxSegmentSize)
		}

		return nil
	}

	err = consumeAndCombineResponses(ctx, c.consumers, resps, c.combiner, sendDiffCb)
	if err != nil {
		return grpcError(err)
	}
	span.AddEvent("consumeAndCombineResponses done")

	// send the final diff if there is anything left
	resp, err := c.combiner.GRPCFinal()
	if err != nil {
		return grpcError(err)
	}
	span.AddEvent("combiner.GRPCFinal() done")

	return c.sendSegmented(req, resp, c.maxSegmentSize)
}

func (c GRPCCollector[T]) sendSegmented(req *http.Request, resp T, maxSize int) error {
	// If no max, then send as-is.