chore: wip 2

Signed-off-by: Nugraha <[email protected]>

Author: ii64

client.go

@@ -65,6 +65,8 @@ func NewClient[Req, Res any](httpClient HTTPClient, url string, options ...Clien
 			EnableGet:        config.EnableGet,
 			GetURLMaxBytes:   config.GetURLMaxBytes,
 			GetUseFallback:   config.GetUseFallback,
+
+			Experimental: config.Experimental,
 		},
 	)
 	if protocolErr != nil {
@@ -213,6 +215,8 @@ type clientConfig struct {
 	GetURLMaxBytes         int
 	GetUseFallback         bool
 	IdempotencyLevel       IdempotencyLevel
+
+	Experimental ExperimentalFeatures
 }
 
 func newClientConfig(rawURL string, options []ClientOption) (*clientConfig, *Error) {

connect_ext_test.go

@@ -2179,16 +2179,20 @@ func TestBidiOverHTTP1(t *testing.T) {
 	client := pingv1connect.NewPingServiceClient(
 		&http.Client{Transport: server.TransportHTTP1()},
 		server.URL(),
+		// connect.WithExperimental(connect.ExperimentalFeatures{
+		// 	AllowBidiStreamOverHTTP11: true,
+		// }),
 	)
 	stream := client.CumSum(context.Background())
 	// Stream creates an async request, can error on Send or Receive.
 	if err := stream.Send(&pingv1.CumSumRequest{Number: 2}); err != nil {
 		assert.ErrorIs(t, err, io.EOF)
 	}
 	_, err := stream.Receive()
-	assert.NotNil(t, err)
 	assert.Equal(t, connect.CodeOf(err), connect.CodeUnknown)
+	assert.NotNil(t, err)
 	assert.Equal(t, err.Error(), "unknown: HTTP status 505 HTTP Version Not Supported")
+
 	assert.Nil(t, stream.CloseRequest())
 	assert.Nil(t, stream.CloseResponse())
 }

duplex_http_call.go

@@ -50,6 +50,8 @@ type duplexHTTPCall struct {
 	responseReady chan struct{}
 	response      *http.Response
 	responseErr   error
+
+	allowBidiStreamOverHTTP11 bool
 }
 
 func newDuplexHTTPCall(
@@ -58,6 +60,7 @@ func newDuplexHTTPCall(
 	url *url.URL,
 	spec Spec,
 	header http.Header,
+	features ExperimentalFeatures,
 ) *duplexHTTPCall {
 	// ensure we make a copy of the url before we pass along to the
 	// Request. This ensures if a transport out of our control wants
@@ -87,6 +90,8 @@ func newDuplexHTTPCall(
 		streamType:    spec.StreamType,
 		request:       request,
 		responseReady: make(chan struct{}),
+
+		allowBidiStreamOverHTTP11: features.AllowBidiStreamOverHTTP11,
 	}
 }
 
@@ -326,7 +331,7 @@ func (d *duplexHTTPCall) makeRequest() {
 		_ = d.CloseWrite()
 		return
 	}
-	if DisallowBidiStreamingHttp11 {
+	if !d.allowBidiStreamOverHTTP11 {
 		if (d.streamType&StreamTypeBidi) == StreamTypeBidi && response.ProtoMajor < 2 {
 			// If we somehow dialed an HTTP/1.x server, fail with an explicit message
 			// rather than returning a more cryptic error later on.

duplex_http_call_test.go

@@ -54,6 +54,7 @@ func TestHTTPCallGetBody(t *testing.T) {
 			serverURL,
 			Spec{StreamType: StreamTypeUnary},
 			http.Header{},
+			ExperimentalFeatures{},
 		)
 		getBodyCalled := false
 		call.onRequestSend = func(*http.Request) {

features.go

@@ -1,5 +1,5 @@
 package connect
 
-var (
-	DisallowBidiStreamingHttp11 = true
-)
+type ExperimentalFeatures struct {
+	AllowBidiStreamOverHTTP11 bool
+}

handler.go

@@ -31,6 +31,7 @@ type Handler struct {
 	protocolHandlers map[string][]protocolHandler // Method to protocol handlers
 	allowMethod      string                       // Allow header
 	acceptPost       string                       // Accept-Post header
+	experimental     ExperimentalFeatures
 }
 
 // NewUnaryHandler constructs a [Handler] for a request-response procedure.
@@ -82,6 +83,8 @@ func NewUnaryHandler[Req, Res any](
 		protocolHandlers: mappedMethodHandlers(protocolHandlers),
 		allowMethod:      sortedAllowMethodValue(protocolHandlers),
 		acceptPost:       sortedAcceptPostValue(protocolHandlers),
+
+		experimental: config.Experimental,
 	}
 }
 
@@ -165,7 +168,7 @@ func (h *Handler) ServeHTTP(responseWriter http.ResponseWriter, request *http.Re
 	if isBidi && request.ProtoMajor < 2 {
 		// Check if we allow bidi stream over HTTP/1.1, enable full-duplex support
 		// and if fail, we fallback to the default behaviour.
-		if !DisallowBidiStreamingHttp11 {
+		if h.experimental.AllowBidiStreamOverHTTP11 {
 			responseController := http.NewResponseController(responseWriter)
 			if err := responseController.EnableFullDuplex(); err == nil {
 				goto Pass
@@ -262,6 +265,8 @@ type handlerConfig struct {
 	ReadMaxBytes                 int
 	SendMaxBytes                 int
 	StreamType                   StreamType
+
+	Experimental ExperimentalFeatures
 }
 
 func newHandlerConfig(procedure string, streamType StreamType, options []HandlerOption) *handlerConfig {
@@ -314,6 +319,7 @@ func (c *handlerConfig) newProtocolHandlers() []protocolHandler {
 			SendMaxBytes:                 c.SendMaxBytes,
 			RequireConnectProtocolHeader: c.RequireConnectProtocolHeader,
 			IdempotencyLevel:             c.IdempotencyLevel,
+			Experimental:                 c.Experimental,
 		}))
 	}
 	return handlers
@@ -333,5 +339,7 @@ func newStreamHandler(
 		protocolHandlers: mappedMethodHandlers(protocolHandlers),
 		allowMethod:      sortedAllowMethodValue(protocolHandlers),
 		acceptPost:       sortedAcceptPostValue(protocolHandlers),
+
+		experimental: config.Experimental,
 	}
 }

option.go

@@ -230,6 +230,10 @@ func WithCodec(codec Codec) Option {
 	return &codecOption{Codec: codec}
 }
 
+func WithExperimental(features ExperimentalFeatures) Option {
+	return &experimentalOption{features}
+}
+
 // WithCompressMinBytes sets a minimum size threshold for compression:
 // regardless of compressor configuration, messages smaller than the configured
 // minimum are sent uncompressed.
@@ -401,6 +405,18 @@ func (o *clientOptionsOption) applyToClient(config *clientConfig) {
 	}
 }
 
+type experimentalOption struct {
+	features ExperimentalFeatures
+}
+
+func (o *experimentalOption) applyToClient(config *clientConfig) {
+	config.Experimental = o.features
+}
+
+func (o *experimentalOption) applyToHandler(config *handlerConfig) {
+	config.Experimental = o.features
+}
+
 type codecOption struct {
 	Codec Codec
 }

protocol.go

@@ -83,6 +83,8 @@ type protocolHandlerParams struct {
 	SendMaxBytes                 int
 	RequireConnectProtocolHeader bool
 	IdempotencyLevel             IdempotencyLevel
+
+	Experimental ExperimentalFeatures
 }
 
 // Handler is the server side of a protocol. HTTP handlers typically support
@@ -132,6 +134,8 @@ type protocolClientParams struct {
 	// The gRPC family of protocols always needs access to a Protobuf codec to
 	// marshal and unmarshal errors.
 	Protobuf Codec
+
+	Experimental ExperimentalFeatures
 }
 
 // Client is the client side of a protocol. HTTP clients typically use a single

protocol_connect.go

@@ -360,7 +360,14 @@ func (c *connectClient) NewConn(
 			} // else effectively unbounded
 		}
 	}
-	duplexCall := newDuplexHTTPCall(ctx, c.HTTPClient, c.URL, spec, header)
+	duplexCall := newDuplexHTTPCall(
+		ctx,
+		c.HTTPClient,
+		c.URL,
+		spec,
+		header,
+		c.Experimental,
+	)
 	var conn streamingClientConn
 	if spec.StreamType == StreamTypeUnary {
 		unaryConn := &connectUnaryClientConn{

protocol_grpc.go

@@ -214,6 +214,7 @@ func (g *grpcHandler) NewConn(
 			},
 			web: g.web,
 		},
+		experimental: g.Experimental,
 	})
 	if failed != nil {
 		// Negotiation failed, so we can't establish a stream.
@@ -280,6 +281,7 @@ func (g *grpcClient) NewConn(
 		g.URL,
 		spec,
 		header,
+		g.Experimental,
 	)
 	conn := &grpcClientConn{
 		spec:             spec,
@@ -466,6 +468,8 @@ type grpcHandlerConn struct {
 	wroteToBody     bool
 	request         *http.Request
 	unmarshaler     grpcUnmarshaler
+
+	experimental ExperimentalFeatures
 }
 
 func (hc *grpcHandlerConn) Spec() Spec {
@@ -515,7 +519,7 @@ func (hc *grpcHandlerConn) Close(err error) (retErr error) {
 		// a well-intentioned client may just not expect the server to be returning
 		// an error for a streaming RPC. Better to accept that we can't always reuse
 		// TCP connections.
-		if DisallowBidiStreamingHttp11 {
+		if !hc.experimental.AllowBidiStreamOverHTTP11 {
 			closeErr := hc.request.Body.Close()
 			if retErr == nil {
 				retErr = closeErr