Connect HTTP Get support

client.go

@@ -59,6 +59,9 @@ func NewClient[Req, Res any](httpClient HTTPClient, url string, options ...Clien
 			BufferPool:       config.BufferPool,
 			ReadMaxBytes:     config.ReadMaxBytes,
 			SendMaxBytes:     config.SendMaxBytes,
+			EnableGet:        config.EnableGet,
+			GetURLMaxBytes:   config.GetURLMaxBytes,
+			GetUseFallback:   config.GetUseFallback,
 		},
 	)
 	if protocolErr != nil {
@@ -182,6 +185,10 @@ type clientConfig struct {
 	BufferPool             *bufferPool
 	ReadMaxBytes           int
 	SendMaxBytes           int
+	EnableGet              bool
+	GetURLMaxBytes         int
+	GetUseFallback         bool
+	IdempotencyLevel       IdempotencyLevel
 }
 
 func newClientConfig(url string, options []ClientOption) (*clientConfig, *Error) {
@@ -224,8 +231,9 @@ func (c *clientConfig) protobuf() Codec {
 
 func (c *clientConfig) newSpec(t StreamType) Spec {
 	return Spec{
-		StreamType: t,
-		Procedure:  c.Procedure,
-		IsClient:   true,
+		StreamType:       t,
+		Procedure:        c.Procedure,
+		IsClient:         true,
+		IdempotencyLevel: c.IdempotencyLevel,
 	}
 }

client_ext_test.go

@@ -19,6 +19,7 @@ import (
 	"errors"
 	"net/http"
 	"net/http/httptest"
+	"strings"
 	"testing"
 
 	"github.com/bufbuild/connect-go"
@@ -89,8 +90,12 @@ func TestClientPeer(t *testing.T) {
 		)
 		ctx := context.Background()
 		// unary
-		_, err := client.Ping(ctx, connect.NewRequest(&pingv1.PingRequest{}))
+		_, err := client.Ping(ctx, connect.NewRequest[pingv1.PingRequest](nil))
 		assert.Nil(t, err)
+		text := strings.Repeat(".", 256)
+		r, err := client.Ping(ctx, connect.NewRequest(&pingv1.PingRequest{Text: text}))
+		assert.Nil(t, err)
+		assert.Equal(t, r.Msg.Text, text)
 		// client streaming
 		clientStream := client.Sum(ctx)
 		t.Cleanup(func() {
@@ -123,6 +128,22 @@ func TestClientPeer(t *testing.T) {
 		t.Parallel()
 		run(t)
 	})
+	t.Run("connect+get", func(t *testing.T) {
+		t.Parallel()
+		run(t,
+			connect.WithHTTPGet(),
+			connect.WithHTTPGetMaxURLSize(256, false),
+			connect.WithSendGzip(),
+		)
+	})
+	t.Run("connect+get fallback", func(t *testing.T) {
+		t.Parallel()
+		run(t,
+			connect.WithHTTPGet(),
+			connect.WithHTTPGetMaxURLSize(1, true),
+			connect.WithSendGzip(),
+		)
+	})
 	t.Run("grpc", func(t *testing.T) {
 		t.Parallel()
 		run(t, connect.WithGRPC())

cmd/protoc-gen-connect-go/main.go

@@ -96,6 +96,17 @@ func main() {
 	)
 }
 
+func needsWithIdempotency(file *protogen.File) bool {
+	for _, service := range file.Services {
+		for _, method := range service.Methods {
+			if methodIdempotency(method) != connect.IdempotencyUnknown {
+				return true
+			}
+		}
+	}
+	return false
+}
+
 func generate(plugin *protogen.Plugin, file *protogen.File) {
 	if len(file.Services) == 0 {
 		return
@@ -139,7 +150,11 @@ func generatePreamble(g *protogen.GeneratedFile, file *protogen.File) {
 		"is not defined, this code was generated with a version of connect newer than the one ",
 		"compiled into your binary. You can fix the problem by either regenerating this code ",
 		"with an older version of connect or updating the connect version compiled into your binary.")
-	g.P("const _ = ", connectPackage.Ident("IsAtLeastVersion0_1_0"))
+	if needsWithIdempotency(file) {
+		g.P("const _ = ", connectPackage.Ident("IsAtLeastVersion1_6_0"))
+	} else {
+		g.P("const _ = ", connectPackage.Ident("IsAtLeastVersion0_1_0"))
+	}
 	g.P()
 }
 
@@ -213,7 +228,17 @@ func generateClientImplementation(g *protogen.GeneratedFile, service *protogen.S
 		)
 		g.P("httpClient,")
 		g.P(`baseURL + "`, procedureName(method), `",`)
-		g.P("opts...,")
+		idempotency := methodIdempotency(method)
+		switch idempotency {
+		case connect.IdempotencyNoSideEffects:
+			g.P(connectPackage.Ident("WithIdempotency"), "(", connectPackage.Ident("IdempotencyNoSideEffects"), "),")
+			g.P(connectPackage.Ident("WithClientOptions"), "(opts...),")
+		case connect.IdempotencyIdempotent:
+			g.P(connectPackage.Ident("WithIdempotency"), "(", connectPackage.Ident("IdempotencyIdempotent"), "),")
+			g.P(connectPackage.Ident("WithClientOptions"), "(opts...),")
+		case connect.IdempotencyUnknown:
+			g.P("opts...,")
+		}
 		g.P("),")
 	}
 	g.P("}")
@@ -327,6 +352,7 @@ func generateServerConstructor(g *protogen.GeneratedFile, service *protogen.Serv
 	for _, method := range service.Methods {
 		isStreamingServer := method.Desc.IsStreamingServer()
 		isStreamingClient := method.Desc.IsStreamingClient()
+		idempotency := methodIdempotency(method)
 		switch {
 		case isStreamingClient && !isStreamingServer:
 			g.P(`mux.Handle("`, procedureName(method), `", `, connectPackage.Ident("NewClientStreamHandler"), "(")
@@ -339,7 +365,16 @@ func generateServerConstructor(g *protogen.GeneratedFile, service *protogen.Serv
 		}
 		g.P(`"`, procedureName(method), `",`)
 		g.P("svc.", method.GoName, ",")
-		g.P("opts...,")
+		switch idempotency {
+		case connect.IdempotencyNoSideEffects:
+			g.P(connectPackage.Ident("WithIdempotency"), "(", connectPackage.Ident("IdempotencyNoSideEffects"), "),")
+			g.P(connectPackage.Ident("WithHandlerOptions"), "(opts...),")
+		case connect.IdempotencyIdempotent:
+			g.P(connectPackage.Ident("WithIdempotency"), "(", connectPackage.Ident("IdempotencyIdempotent"), "),")
+			g.P(connectPackage.Ident("WithHandlerOptions"), "(opts...),")
+		case connect.IdempotencyUnknown:
+			g.P("opts...,")
+		}
 		g.P("))")
 	}
 	g.P(`return "/`, reflectionName(service), `/", mux`)
@@ -433,6 +468,22 @@ func isDeprecatedMethod(method *protogen.Method) bool {
 	return ok && methodOptions.GetDeprecated()
 }
 
+func methodIdempotency(method *protogen.Method) connect.IdempotencyLevel {
+	methodOptions, ok := method.Desc.Options().(*descriptorpb.MethodOptions)
+	if !ok {
+		return connect.IdempotencyUnknown
+	}
+	switch methodOptions.GetIdempotencyLevel() {
+	case descriptorpb.MethodOptions_NO_SIDE_EFFECTS:
+		return connect.IdempotencyNoSideEffects
+	case descriptorpb.MethodOptions_IDEMPOTENT:
+		return connect.IdempotencyIdempotent
+	case descriptorpb.MethodOptions_IDEMPOTENCY_UNKNOWN:
+		return connect.IdempotencyUnknown
+	}
+	return connect.IdempotencyUnknown
+}
+
 // Raggedy comments in the generated code are driving me insane. This
 // word-wrapping function is ruinously inefficient, but it gets the job done.
 func wrapComments(g *protogen.GeneratedFile, elems ...any) {

codec.go

@@ -15,6 +15,8 @@
 package connect
 
 import (
+	"bytes"
+	"encoding/json"
 	"fmt"
 
 	"google.golang.org/protobuf/encoding/protojson"
@@ -49,6 +51,32 @@ type Codec interface {
 	Unmarshal([]byte, any) error
 }
 
+// stableCodec is an extension to Codec for serializing with stable output.
+type stableCodec interface {
+	Codec
+
+	// MarshalStable marshals the given message with stable field ordering.
+	//
+	// MarshalStable should return the same output for a given input. Although
+	// it is not guaranteed to be canonicalized, the marshalling routine for
+	// MarshalStable will opt for the most normalized output available for a
+	// given serialization.
+	//
+	// For practical reasons, it is possible for MarshalStable to return two
+	// different results for two inputs considered to be "equal" in their own
+	// domain, and it may change in the future with codec updates, but for
+	// any given concrete value and any given version, it should return the
+	// same output.
+	MarshalStable(any) ([]byte, error)
+
+	// IsBinary returns true if the marshalled data is binary for this codec.
+	//
+	// If this function returns false, the data returned from Marshal and
+	// MarshalStable are considered valid text and may be used in contexts
+	// where text is expected.
+	IsBinary() bool
+}
+
 type protoBinaryCodec struct{}
 
 var _ Codec = (*protoBinaryCodec)(nil)
@@ -71,6 +99,24 @@ func (c *protoBinaryCodec) Unmarshal(data []byte, message any) error {
 	return proto.Unmarshal(data, protoMessage)
 }
 
+func (c *protoBinaryCodec) MarshalStable(message any) ([]byte, error) {
+	protoMessage, ok := message.(proto.Message)
+	if !ok {
+		return nil, errNotProto(message)
+	}
+	// protobuf does not offer a canonical output today, so this format is not
+	// guaranteed to match deterministic output from other protobuf libraries.
+	// In addition, unknown fields may cause inconsistent output for otherwise
+	// equal messages.
+	// https://github.com/golang/protobuf/issues/1121
+	options := proto.MarshalOptions{Deterministic: true}
+	return options.Marshal(protoMessage)
+}
+
+func (c *protoBinaryCodec) IsBinary() bool {
+	return true
+}
+
 type protoJSONCodec struct {
 	name string
 }
@@ -97,6 +143,27 @@ func (c *protoJSONCodec) Unmarshal(binary []byte, message any) error {
 	return options.Unmarshal(binary, protoMessage)
 }
 
+func (c *protoJSONCodec) MarshalStable(message any) ([]byte, error) {
+	// protojson does not offer a "deterministic" field ordering, but fields
+	// are still ordered consistently by their index. However, protojson can
+	// output inconsistent whitespace for some reason, therefore it is
+	// suggested to use a formatter to ensure consistent formatting.
+	// https://github.com/golang/protobuf/issues/1373
+	compactedJSON := new(bytes.Buffer)
+	messageJSON, err := c.Marshal(message)
+	if err != nil {
+		return nil, err
+	}
+	if err = json.Compact(compactedJSON, messageJSON); err != nil {
+		return nil, err
+	}
+	return compactedJSON.Bytes(), nil
+}
+
+func (c *protoJSONCodec) IsBinary() bool {
+	return false
+}
+
 // readOnlyCodecs is a read-only interface to a map of named codecs.
 type readOnlyCodecs interface {
 	// Get gets the Codec with the given name.

codec_test.go

@@ -0,0 +1,95 @@
+// Copyright 2021-2023 Buf Technologies, Inc.
+//
+// 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 connect
+
+import (
+	"bytes"
+	"testing"
+	"testing/quick"
+
+	pingv1 "github.com/bufbuild/connect-go/internal/gen/connect/ping/v1"
+	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/types/known/structpb"
+)
+
+func convertMapToInterface(stringMap map[string]string) map[string]interface{} {
+	interfaceMap := make(map[string]interface{})
+	for key, value := range stringMap {
+		interfaceMap[key] = value
+	}
+	return interfaceMap
+}
+
+func TestCodecRoundTrips(t *testing.T) {
+	t.Parallel()
+	makeRoundtrip := func(codec Codec) func(string, int64) bool {
+		return func(text string, number int64) bool {
+			got := pingv1.PingRequest{}
+			want := pingv1.PingRequest{Text: text, Number: number}
+			data, err := codec.Marshal(&want)
+			if err != nil {
+				t.Fatal(err)
+			}
+			err = codec.Unmarshal(data, &got)
+			if err != nil {
+				t.Fatal(err)
+			}
+			return proto.Equal(&got, &want)
+		}
+	}
+	if err := quick.Check(makeRoundtrip(&protoBinaryCodec{}), nil /* config */); err != nil {
+		t.Error(err)
+	}
+	if err := quick.Check(makeRoundtrip(&protoJSONCodec{}), nil /* config */); err != nil {
+		t.Error(err)
+	}
+}
+
+func TestStableCodec(t *testing.T) {
+	t.Parallel()
+	makeRoundtrip := func(codec stableCodec) func(map[string]string) bool {
+		return func(input map[string]string) bool {
+			initialProto, err := structpb.NewStruct(convertMapToInterface(input))
+			if err != nil {
+				t.Fatal(err)
+			}
+			want, err := codec.MarshalStable(initialProto)
+			if err != nil {
+				t.Fatal(err)
+			}
+			for i := 0; i < 10; i++ {
+				roundtripProto := &structpb.Struct{}
+				err = codec.Unmarshal(want, roundtripProto)
+				if err != nil {
+					t.Fatal(err)
+				}
+				got, err := codec.MarshalStable(roundtripProto)
+				if err != nil {
+					t.Fatal(err)
+				}
+				if !bytes.Equal(got, want) {
+					return false
+				}
+			}
+			return true
+		}
+	}
+	if err := quick.Check(makeRoundtrip(&protoBinaryCodec{}), nil /* config */); err != nil {
+		t.Error(err)
+	}
+	if err := quick.Check(makeRoundtrip(&protoJSONCodec{}), nil /* config */); err != nil {
+		t.Error(err)
+	}
+}