Support sending raw HTTP request in reference client (#737)

This still expects the test case to include a normal request, too. That
allows us to continue using the "expected response" computation, based
on the request. But we can then override the actual HTTP request, to
simulate some behavior (mainly around on-the-wire encoding) that the
reference client doesn't usually exhibit. The main things that are _not_
possible with raw requests are the ability to do cancellations or to
introduce delays between writes.

internal/app/referenceclient/client.go

@@ -204,6 +204,10 @@ func invoke(ctx context.Context, req *v1.ClientCompatRequest) (*v1.ClientRespons
 		return nil, errors.New("an HTTP version must be specified")
 	}
 
+	if req.RawRequest != nil {
+		transport = &rawRequestSender{transport: transport, rawRequest: req.RawRequest}
+	}
+
 	// Create client options based on protocol of the implementation
 	clientOptions := []connect.ClientOption{connect.WithHTTPGet()}
 	switch req.Protocol {

internal/app/referenceclient/raw_request.go

@@ -0,0 +1,110 @@
+// Copyright 2023 The Connect Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package referenceclient
+
+import (
+	"bytes"
+	"encoding/base64"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"strconv"
+
+	"connectrpc.com/conformance/internal"
+	conformancev1 "connectrpc.com/conformance/internal/gen/proto/go/connectrpc/conformance/v1"
+)
+
+type rawRequestSender struct {
+	transport  http.RoundTripper
+	rawRequest *conformancev1.RawHTTPRequest
+}
+
+func (r *rawRequestSender) RoundTrip(orig *http.Request) (*http.Response, error) {
+	switch r.rawRequest.Body.(type) {
+	case *conformancev1.RawHTTPRequest_Unary, *conformancev1.RawHTTPRequest_Stream, nil:
+	default:
+		return nil, fmt.Errorf("raw request has invalid body definition: %T", r.rawRequest.Body)
+	}
+
+	uri := r.rawRequest.Uri
+	if len(r.rawRequest.RawQueryParams) > 0 || len(r.rawRequest.EncodedQueryParams) > 0 {
+		reqURL, err := url.Parse(r.rawRequest.Uri)
+		if err != nil {
+			return nil, fmt.Errorf("raw request has invalid URI: %s: %w", r.rawRequest.Uri, err)
+		}
+		vals := reqURL.Query()
+		for _, param := range r.rawRequest.RawQueryParams {
+			vals[param.Name] = append(vals[param.Name], param.Value...)
+		}
+		for _, param := range r.rawRequest.EncodedQueryParams {
+			var buf bytes.Buffer
+			if err := internal.WriteRawMessageContents(param.Value, &buf); err != nil {
+				return nil, fmt.Errorf("raw request has invalid encoded query param %s: %w", param.Name, err)
+			}
+			var paramVal string
+			if param.Base64Encode {
+				paramVal = base64.URLEncoding.EncodeToString(buf.Bytes())
+			} else {
+				paramVal = buf.String()
+			}
+			vals[param.Name] = append(vals[param.Name], paramVal)
+		}
+		reqURL.RawQuery = vals.Encode()
+		uri = reqURL.String()
+	}
+
+	pipeReader, pipeWriter := io.Pipe()
+	req, err := http.NewRequestWithContext(
+		orig.Context(),
+		r.rawRequest.Verb,
+		fmt.Sprintf("%s://%s%s", orig.URL.Scheme, orig.URL.Host, uri),
+		pipeReader,
+	)
+	if err != nil {
+		if orig.Body != nil {
+			_ = orig.Body.Close()
+		}
+		return nil, err
+	}
+
+	internal.AddHeaders(r.rawRequest.Headers, req.Header)
+	if length, err := strconv.Atoi(req.Header.Get("Content-Length")); err == nil {
+		req.ContentLength = int64(length)
+	}
+
+	// Write the request body to the pipe.
+	go func() {
+		defer func() {
+			_ = pipeWriter.Close()
+		}()
+		switch contents := r.rawRequest.Body.(type) {
+		case *conformancev1.RawHTTPRequest_Unary:
+			_ = internal.WriteRawMessageContents(contents.Unary, pipeWriter)
+		case *conformancev1.RawHTTPRequest_Stream:
+			_ = internal.WriteRawStreamContents(contents.Stream, pipeWriter)
+		}
+	}()
+
+	// Drain the original request body and close it.
+	go func() {
+		defer func() {
+			_ = orig.Body.Close()
+		}()
+		_, _ = io.Copy(io.Discard, orig.Body)
+	}()
+
+	return r.transport.RoundTrip(req)
+}