rpc: Support handlers of any arity (#4630)

Previously, we had of hand written adapters that allowed exposing go
functions over our JSON-RPC boundary.  Changes to the number of
arguments or return values would require changes to the adapter used
(or writing a new adapter by copy/pasting an existing one).

Now, we have a single `NewHandler` which allows exposing arbitrary go
functions over the RPC boundary (as long as the follow some rules,
like taking a `context.Context` as the first paremeter, and returning
a value and `error` or just an `error`.

As part of this change, the `IObserver` type is now called `Observer`
and it is passed by reference instead of value in signatures.

cli/azd/internal/vsrpc/aspire_service.go

@@ -28,7 +28,7 @@ func newAspireService(server *Server) *aspireService {
 // GetAspireHostAsync is the server implementation of:
 // ValueTask<AspireHost> GetAspireHostAsync(Session session, string aspireEnv, CancellationToken cancellationToken).
 func (s *aspireService) GetAspireHostAsync(
-	ctx context.Context, rc RequestContext, aspireEnv string, observer IObserver[ProgressMessage],
+	ctx context.Context, rc RequestContext, aspireEnv string, observer *Observer[ProgressMessage],
 ) (*AspireHost, error) {
 	session, err := s.server.validateSession(rc.Session)
 	if err != nil {
@@ -65,7 +65,7 @@ func (s *aspireService) GetAspireHostAsync(
 // RenameAspireHostAsync is the server implementation of:
 // ValueTask RenameAspireHostAsync(Session session, string newPath, CancellationToken cancellationToken).
 func (s *aspireService) RenameAspireHostAsync(
-	ctx context.Context, rc RequestContext, newPath string, observer IObserver[ProgressMessage],
+	ctx context.Context, rc RequestContext, newPath string, observer *Observer[ProgressMessage],
 ) error {
 	_, err := s.server.validateSession(rc.Session)
 	if err != nil {
@@ -79,7 +79,7 @@ func (s *aspireService) RenameAspireHostAsync(
 // ServeHTTP implements http.Handler.
 func (s *aspireService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	serveRpc(w, r, map[string]Handler{
-		"GetAspireHostAsync":    HandlerFunc3(s.GetAspireHostAsync),
-		"RenameAspireHostAsync": HandlerAction3(s.RenameAspireHostAsync),
+		"GetAspireHostAsync":    NewHandler(s.GetAspireHostAsync),
+		"RenameAspireHostAsync": NewHandler(s.RenameAspireHostAsync),
 	})
 }

cli/azd/internal/vsrpc/debug_service.go

@@ -53,7 +53,7 @@ func (s *debugService) TestCancelAsync(ctx context.Context, timeoutMs int) (bool
 // ValueTask<bool> TestIObserverAsync(int, CancellationToken);
 //
 // It emits a sequence of integers to the observer, from 0 to max, and then completes the observer, before returning.
-func (s *debugService) TestIObserverAsync(ctx context.Context, max int, observer IObserver[int]) error {
+func (s *debugService) TestIObserverAsync(ctx context.Context, max int, observer *Observer[int]) error {
 	for i := 0; i < max; i++ {
 		_ = observer.OnNext(ctx, i)
 	}
@@ -107,9 +107,9 @@ func (s *debugService) FetchTokenAsync(ctx context.Context, sessionId Session) (
 // ServeHTTP implements http.Handler.
 func (s *debugService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	serveRpc(w, r, map[string]Handler{
-		"TestCancelAsync":    HandlerFunc1(s.TestCancelAsync),
-		"TestIObserverAsync": HandlerAction2(s.TestIObserverAsync),
-		"TestPanicAsync":     HandlerAction1(s.TestPanicAsync),
-		"FetchTokenAsync":    HandlerFunc1(s.FetchTokenAsync),
+		"TestCancelAsync":    NewHandler(s.TestCancelAsync),
+		"TestIObserverAsync": NewHandler(s.TestIObserverAsync),
+		"TestPanicAsync":     NewHandler(s.TestPanicAsync),
+		"FetchTokenAsync":    NewHandler(s.FetchTokenAsync),
 	})
 }

cli/azd/internal/vsrpc/environment_service.go

@@ -32,7 +32,7 @@ func newEnvironmentService(server *Server) *environmentService {
 // ValueTask<IEnumerable<EnvironmentInfo>> GetEnvironmentsAsync(
 // RequestContext, IObserver<ProgressMessage>, CancellationToken);
 func (s *environmentService) GetEnvironmentsAsync(
-	ctx context.Context, rc RequestContext, observer IObserver[ProgressMessage],
+	ctx context.Context, rc RequestContext, observer *Observer[ProgressMessage],
 ) ([]*EnvironmentInfo, error) {
 	session, err := s.server.validateSession(rc.Session)
 	if err != nil {
@@ -71,7 +71,7 @@ func (s *environmentService) GetEnvironmentsAsync(
 // SetCurrentEnvironmentAsync is the server implementation of:
 // ValueTask<bool> SetCurrentEnvironmentAsync(RequestContext, string, IObserver<ProgressMessage>, CancellationToken);
 func (s *environmentService) SetCurrentEnvironmentAsync(
-	ctx context.Context, rc RequestContext, name string, observer IObserver[ProgressMessage],
+	ctx context.Context, rc RequestContext, name string, observer *Observer[ProgressMessage],
 ) (bool, error) {
 	session, err := s.server.validateSession(rc.Session)
 	if err != nil {
@@ -107,7 +107,7 @@ const (
 // DeleteEnvironmentAsync is the server implementation of:
 // ValueTask<bool> DeleteEnvironmentAsync(RequestContext, string, IObserver<ProgressMessage>, int, CancellationToken);
 func (s *environmentService) DeleteEnvironmentAsync(
-	ctx context.Context, rc RequestContext, name string, mode int, observer IObserver[ProgressMessage],
+	ctx context.Context, rc RequestContext, name string, mode int, observer *Observer[ProgressMessage],
 ) (bool, error) {
 	session, err := s.server.validateSession(rc.Session)
 	if err != nil {
@@ -196,13 +196,13 @@ func (s *environmentService) DeleteEnvironmentAsync(
 // ServeHTTP implements http.Handler.
 func (s *environmentService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	serveRpc(w, r, map[string]Handler{
-		"CreateEnvironmentAsync":     HandlerFunc3(s.CreateEnvironmentAsync),
-		"GetEnvironmentsAsync":       HandlerFunc2(s.GetEnvironmentsAsync),
-		"LoadEnvironmentAsync":       HandlerFunc3(s.LoadEnvironmentAsync),
-		"OpenEnvironmentAsync":       HandlerFunc3(s.OpenEnvironmentAsync),
-		"SetCurrentEnvironmentAsync": HandlerFunc3(s.SetCurrentEnvironmentAsync),
-		"DeleteEnvironmentAsync":     HandlerFunc4(s.DeleteEnvironmentAsync),
-		"RefreshEnvironmentAsync":    HandlerFunc3(s.RefreshEnvironmentAsync),
-		"DeployAsync":                HandlerFunc3(s.DeployAsync),
+		"CreateEnvironmentAsync":     NewHandler(s.CreateEnvironmentAsync),
+		"GetEnvironmentsAsync":       NewHandler(s.GetEnvironmentsAsync),
+		"LoadEnvironmentAsync":       NewHandler(s.LoadEnvironmentAsync),
+		"OpenEnvironmentAsync":       NewHandler(s.OpenEnvironmentAsync),
+		"SetCurrentEnvironmentAsync": NewHandler(s.SetCurrentEnvironmentAsync),
+		"DeleteEnvironmentAsync":     NewHandler(s.DeleteEnvironmentAsync),
+		"RefreshEnvironmentAsync":    NewHandler(s.RefreshEnvironmentAsync),
+		"DeployAsync":                NewHandler(s.DeployAsync),
 	})
 }

cli/azd/internal/vsrpc/environment_service_create.go

@@ -18,7 +18,7 @@ import (
 // CreateEnvironmentAsync is the server implementation of:
 // ValueTask<bool> CreateEnvironmentAsync(RequestContext, Environment, IObserver<ProgressMessage>, CancellationToken);
 func (s *environmentService) CreateEnvironmentAsync(
-	ctx context.Context, rc RequestContext, newEnv Environment, observer IObserver[ProgressMessage],
+	ctx context.Context, rc RequestContext, newEnv Environment, observer *Observer[ProgressMessage],
 ) (bool, error) {
 	session, err := s.server.validateSession(rc.Session)
 	if err != nil {

cli/azd/internal/vsrpc/environment_service_deploy.go

@@ -17,7 +17,7 @@ import (
 //
 // While it is named simply `DeployAsync`, it behaves as if the user had run `azd provision` and `azd deploy`.
 func (s *environmentService) DeployAsync(
-	ctx context.Context, rc RequestContext, name string, observer IObserver[ProgressMessage],
+	ctx context.Context, rc RequestContext, name string, observer *Observer[ProgressMessage],
 ) (*Environment, error) {
 	session, err := s.server.validateSession(rc.Session)
 	if err != nil {

cli/azd/internal/vsrpc/environment_service_load.go

@@ -21,7 +21,7 @@ import (
 // already cached) and is faster than `LoadEnvironmentAsync` in cases where we have not cached the manifest. This means
 // the Services array of the returned environment may be empty.
 func (s *environmentService) OpenEnvironmentAsync(
-	ctx context.Context, rc RequestContext, name string, observer IObserver[ProgressMessage],
+	ctx context.Context, rc RequestContext, name string, observer *Observer[ProgressMessage],
 ) (*Environment, error) {
 	session, err := s.server.validateSession(rc.Session)
 	if err != nil {
@@ -43,7 +43,7 @@ func (s *environmentService) OpenEnvironmentAsync(
 // the environment (like service endpoints) may not be available. Use `RefreshEnvironmentAsync` to load the environment and
 // fetch information from Azure.
 func (s *environmentService) LoadEnvironmentAsync(
-	ctx context.Context, rc RequestContext, name string, observer IObserver[ProgressMessage],
+	ctx context.Context, rc RequestContext, name string, observer *Observer[ProgressMessage],
 ) (*Environment, error) {
 	session, err := s.server.validateSession(rc.Session)
 	if err != nil {

cli/azd/internal/vsrpc/environment_service_refresh.go

@@ -25,7 +25,7 @@ import (
 // to accept some loss of information in favor of a faster load time, use `LoadEnvironmentAsync` instead, which does not
 // contact azure to compute service endpoints or last deployment information.
 func (s *environmentService) RefreshEnvironmentAsync(
-	ctx context.Context, rc RequestContext, name string, observer IObserver[ProgressMessage],
+	ctx context.Context, rc RequestContext, name string, observer *Observer[ProgressMessage],
 ) (*Environment, error) {
 	session, err := s.server.validateSession(rc.Session)
 	if err != nil {
@@ -41,7 +41,7 @@ func (s *environmentService) RefreshEnvironmentAsync(
 }
 
 func (s *environmentService) refreshEnvironmentAsync(
-	ctx context.Context, container *container, name string, observer IObserver[ProgressMessage],
+	ctx context.Context, container *container, name string, observer *Observer[ProgressMessage],
 ) (*Environment, error) {
 	env, err := s.loadEnvironmentAsync(ctx, container, name, true)
 	if err != nil {