Support dynamic messages with Initializer options

[emcfarlane]

To support dynamic clients and servers we need to initialize messages on Receive. This adds two new options WithRequestInitializer and WithResponseInitializer that provide InitializerFuncs that help construct request messages before use. Proto based schemas can use the new Schema field on Spec to introspect the method types. Other IDLs can set their own schema using the WithSchema option.

Dynamic Message Initializer

As an example to support dynamicpb.Message we will inspect the schema and set the message descriptor. An intializer func can be run on the client or handller. This func is provided to the examples below as an option:

// dynamicMessageInitialzier inits a dynamicpb.Message descriptor to the correct
// input or output type for the method.
func dynamicMessageInitializer(spec Spec, msg any) error {
	dynamic, ok := msg.(*dynamicpb.Message)
	if !ok {
		return nil
	}
	desc, ok := spec.Schema.(protoreflect.MethodDescriptor)
	if !ok {
		return fmt.Errorf("unexpected schema type %T for %T message", spec.Schema, dynamic)
	}
	// Client initializer is run on response, Server on request.
	if spec.IsClient {
		*dynamic = *dynamicpb.NewMessage(desc.Output())
	} else {
		*dynamic = *dynamicpb.NewMessage(desc.Input())
	}
	return nil
}

Dynamic Clients

Construct a client using dynamicpb.Message. Ensure the URL includes the suffix of the handler method to invoke and to include WithSchema and WithResponseInitializer options:

desc, _ = protoregistry.GlobalFiles.FindDescriptorByName("connect.ping.v1.PingService.Ping")
methodDesc := desc.(protoreflect.MethodDescriptor)
// Create a client with [dynamicpb.Message] as the request and response message types.
client := connect.NewClient[dynamicpb.Message, dynamicpb.Message](
	http.DefaultClient,
	serverURL + "/connect.ping.v1.PingService/Ping",
	connect.WithSchema(methodDesc),
	connect.WithResponseInitializer(dynamicMessageInitializer),
	connect.WithIdempotency(connect.IdempotencyNoSideEffects),
)
// Build the message using the input descriptor.
msg := dynamicpb.NewMessage(methodDesc.Input())
msg.Set(
	methodDesc.Input().Fields().ByName("number"),
	protoreflect.ValueOfInt64(42),
)
res, _ := client.CallUnary(context.Background(), connect.NewRequest(msg))
// Get values from the message using the output descriptor.
got := res.Msg.Get(
	methodDesc.Output().Fields().ByName("number")
).Int()

Dynamic Handlers

Create the server func with a signature that includes the dynamicpb.Message. Then construct the handler specifying the procedure URL and ensure to include WithSchema and WithRequestInitializer options:

// Create a handler func with [dynamicpb.Message] as the request and response message types.
dynamicPing := func(_ context.Context, req *connect.Request[dynamicpb.Message]) (*connect.Response[dynamicpb.Message], error) {
	got := req.Msg.Get(methodDesc.Input().Fields().ByName("number")).Int()
	msg := dynamicpb.NewMessage(methodDesc.Output())
	msg.Set(
		methodDesc.Output().Fields().ByName("number"),
		protoreflect.ValueOfInt64(got),
	)
	return connect.NewResponse(msg), nil
}
mux := http.NewServeMux()
mux.Handle("/connect.ping.v1.PingService/Ping",
	connect.NewUnaryHandler(
		"/connect.ping.v1.PingService/Ping",
		dynamicPing,
		connect.WithSchema(methodDesc),
		connect.WithRequestInitializer(dynamicMessageInitializer),
		connect.WithIdempotency(connect.IdempotencyNoSideEffects),
	),
)

Fixes #523

[jhump]

FYI, I've merged the schema PR, so this needs a rebase

[akshayjshah]

Haven't looked at the implementation in detail, but the exported API makes sense to me.

[emcfarlane]

Closing, reopening as a fork PR: #640