Method Forward

backend/go/cloud-proxy/proxy.go:268–390

Forward shovels bytes between a Forward gRPC stream and an upstream HTTP request. First request message carries path/method/headers and the initial body chunk; subsequent messages append body chunks. The first reply carries upstream status + response headers; subsequent replies stream body chunks un

(ctx context.Context, in <-chan *pb.ForwardRequest, out chan<- *pb.ForwardReply)

// Cancellation of ctx (the gRPC stream context) closes the upstream
// connection.
func (c *CloudProxy) Forward(ctx context.Context, in <-chan *pb.ForwardRequest, out chan<- *pb.ForwardReply) error {
	defer close(out)

	cfg := c.cfg.Load()
	if cfg == nil {
		return grpcerrors.ModelNotLoaded("cloud-proxy")
	}
	if cfg.mode != modePassthrough {
		return fmt.Errorf("cloud-proxy: Forward only valid in passthrough mode (have %s)", cfg.mode)
	}

	first, ok := <-in
	if !ok {
		return errors.New("cloud-proxy: Forward stream closed before first request")
	}

	// Honour the per-request path only when the configured upstream_url
	// has no path of its own — gallery convention is to put the
	// canonical path in upstream_url.
	fullURL, err := composeURL(cfg.upstreamURL, first.GetPath())
	if err != nil {
		return err
	}

	method := first.GetMethod()
	if method == "" {
		method = http.MethodPost
	}

	// Pipe the body in from the gRPC stream so the HTTP request can
	// start before the client finishes sending. The pipe-reader is
	// closed via CloseWithError on the error paths so the writer
	// goroutine doesn't block forever.
	pr, pw := io.Pipe()

	go func() {
		var writeErr error
		defer func() { _ = pw.CloseWithError(writeErr) }()
		if len(first.GetBodyChunk()) > 0 {
			if _, writeErr = pw.Write(first.GetBodyChunk()); writeErr != nil {
				return
			}
		}
		for req := range in {
			if len(req.GetBodyChunk()) == 0 {
				continue
			}
			if _, writeErr = pw.Write(req.GetBodyChunk()); writeErr != nil {
				return
			}
		}
	}()

	req, err := http.NewRequestWithContext(ctx, method, fullURL, pr)
	if err != nil {
		_ = pr.CloseWithError(err) // unblocks the body-pump's pw.Write
		return fmt.Errorf("cloud-proxy: build request: %w", err)
	}