feat: support http->http in gateway (#4605)

gateway/server.go

@@ -3,22 +3,29 @@ package gateway
 import (
 	"context"
 	"fmt"
+	"io"
 	"net/http"
+	"net/url"
 	"strings"
+	"time"
 
 	"github.com/fullstorydev/grpcurl"
 	"github.com/golang/protobuf/jsonpb"
 	"github.com/jhump/protoreflect/grpcreflect"
+	"github.com/zeromicro/go-zero/core/logc"
 	"github.com/zeromicro/go-zero/core/logx"
 	"github.com/zeromicro/go-zero/core/mr"
 	"github.com/zeromicro/go-zero/core/threading"
 	"github.com/zeromicro/go-zero/gateway/internal"
 	"github.com/zeromicro/go-zero/rest"
+	"github.com/zeromicro/go-zero/rest/httpc"
 	"github.com/zeromicro/go-zero/rest/httpx"
 	"github.com/zeromicro/go-zero/zrpc"
 	"google.golang.org/grpc/codes"
 )
 
+const defaultHttpScheme = "http"
+
 type (
 	// Server is a gateway server.
 	Server struct {
@@ -83,52 +90,11 @@ func (s *Server) build() error {
 			source <- up
 		}
 	}, func(up Upstream, writer mr.Writer[rest.Route], cancel func(error)) {
-		var cli zrpc.Client
-		if s.dialer != nil {
-			cli = s.dialer(up.Grpc)
-		} else {
-			cli = zrpc.MustNewClient(up.Grpc)
-		}
-		s.conns = append(s.conns, cli)
-
-		source, err := s.createDescriptorSource(cli, up)
-		if err != nil {
-			cancel(fmt.Errorf("%s: %w", up.Name, err))
-			return
-		}
-
-		methods, err := internal.GetMethods(source)
-		if err != nil {
-			cancel(fmt.Errorf("%s: %w", up.Name, err))
-			return
-		}
-
-		resolver := grpcurl.AnyResolverFromDescriptorSource(source)
-		for _, m := range methods {
-			if len(m.HttpMethod) > 0 && len(m.HttpPath) > 0 {
-				writer.Write(rest.Route{
-					Method:  m.HttpMethod,
-					Path:    m.HttpPath,
-					Handler: s.buildHandler(source, resolver, cli, m.RpcPath),
-				})
-			}
-		}
-
-		methodSet := make(map[string]struct{})
-		for _, m := range methods {
-			methodSet[m.RpcPath] = struct{}{}
-		}
-		for _, m := range up.Mappings {
-			if _, ok := methodSet[m.RpcPath]; !ok {
-				cancel(fmt.Errorf("%s: rpc method %s not found", up.Name, m.RpcPath))
-				return
-			}
-
-			writer.Write(rest.Route{
-				Method:  strings.ToUpper(m.Method),
-				Path:    m.Path,
-				Handler: s.buildHandler(source, resolver, cli, m.RpcPath),
-			})
+		// up.Grpc and up.Http are exclusive
+		if up.Grpc != nil {
+			s.buildGrpcRoute(up, writer, cancel)
+		} else if up.Http != nil {
+			s.buildHttpRoute(up, writer)
 		}
 	}, func(pipe <-chan rest.Route, cancel func(error)) {
 		for route := range pipe {
@@ -137,7 +103,7 @@ func (s *Server) build() error {
 	})
 }
 
-func (s *Server) buildHandler(source grpcurl.DescriptorSource, resolver jsonpb.AnyResolver,
+func (s *Server) buildGrpcHandler(source grpcurl.DescriptorSource, resolver jsonpb.AnyResolver,
 	cli zrpc.Client, rpcPath string) func(http.ResponseWriter, *http.Request) {
 	return func(w http.ResponseWriter, r *http.Request) {
 		parser, err := internal.NewRequestParser(r, resolver)
@@ -160,31 +126,119 @@ func (s *Server) buildHandler(source grpcurl.DescriptorSource, resolver jsonpb.A
 	}
 }
 
-func (s *Server) createDescriptorSource(cli zrpc.Client, up Upstream) (grpcurl.DescriptorSource, error) {
-	var source grpcurl.DescriptorSource
-	var err error
-
-	if len(up.ProtoSets) > 0 {
-		source, err = grpcurl.DescriptorSourceFromProtoSets(up.ProtoSets...)
-		if err != nil {
-			return nil, err
-		}
+func (s *Server) buildGrpcRoute(up Upstream, writer mr.Writer[rest.Route], cancel func(error)) {
+	var cli zrpc.Client
+	if s.dialer != nil {
+		cli = s.dialer(*up.Grpc)
 	} else {
-		client := grpcreflect.NewClientAuto(context.Background(), cli.Conn())
-		source = grpcurl.DescriptorSourceFromServer(context.Background(), client)
+		cli = zrpc.MustNewClient(*up.Grpc)
+	}
+	s.conns = append(s.conns, cli)
+
+	source, err := createDescriptorSource(cli, up)
+	if err != nil {
+		cancel(fmt.Errorf("%s: %w", up.Name, err))
+		return
 	}
 
-	return source, nil
+	methods, err := internal.GetMethods(source)
+	if err != nil {
+		cancel(fmt.Errorf("%s: %w", up.Name, err))
+		return
+	}
+
+	resolver := grpcurl.AnyResolverFromDescriptorSource(source)
+	for _, m := range methods {
+		if len(m.HttpMethod) > 0 && len(m.HttpPath) > 0 {
+			writer.Write(rest.Route{
+				Method:  m.HttpMethod,
+				Path:    m.HttpPath,
+				Handler: s.buildGrpcHandler(source, resolver, cli, m.RpcPath),
+			})
+		}
+	}
+
+	methodSet := make(map[string]struct{})
+	for _, m := range methods {
+		methodSet[m.RpcPath] = struct{}{}
+	}
+	for _, m := range up.Mappings {
+		if _, ok := methodSet[m.RpcPath]; !ok {
+			cancel(fmt.Errorf("%s: rpc method %s not found", up.Name, m.RpcPath))
+			return
+		}
+
+		writer.Write(rest.Route{
+			Method:  strings.ToUpper(m.Method),
+			Path:    m.Path,
+			Handler: s.buildGrpcHandler(source, resolver, cli, m.RpcPath),
+		})
+	}
+}
+
+func (s *Server) buildHttpHandler(target *HttpClientConf) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set(httpx.ContentType, httpx.JsonContentType)
+		req, err := buildRequestWithNewTarget(r, target)
+		if err != nil {
+			httpx.ErrorCtx(r.Context(), w, err)
+			return
+		}
+
+		if target.Timeout > 0 {
+			timeout := time.Duration(target.Timeout) * time.Millisecond
+			ctx, cancel := context.WithTimeout(r.Context(), timeout)
+			defer cancel()
+			req = req.WithContext(ctx)
+		}
+
+		resp, err := httpc.DoRequest(req)
+		if err != nil {
+			httpx.ErrorCtx(r.Context(), w, err)
+			return
+		}
+		defer resp.Body.Close()
+
+		for key, values := range resp.Header {
+			for _, value := range values {
+				w.Header().Add(key, value)
+			}
+		}
+
+		w.WriteHeader(resp.StatusCode)
+		if _, err = io.Copy(w, resp.Body); err != nil {
+			// log the error with original request info
+			logc.Error(r.Context(), err)
+		}
+	}
+}
+
+func (s *Server) buildHttpRoute(up Upstream, writer mr.Writer[rest.Route]) {
+	for _, m := range up.Mappings {
+		writer.Write(rest.Route{
+			Method:  strings.ToUpper(m.Method),
+			Path:    m.Path,
+			Handler: s.buildHttpHandler(up.Http),
+		})
+	}
 }
 
 func (s *Server) ensureUpstreamNames() error {
 	for i := 0; i < len(s.upstreams); i++ {
-		target, err := s.upstreams[i].Grpc.BuildTarget()
-		if err != nil {
-			return err
+		if len(s.upstreams[i].Name) > 0 {
+			continue
 		}
 
-		s.upstreams[i].Name = target
+		if s.upstreams[i].Grpc != nil {
+			target, err := s.upstreams[i].Grpc.BuildTarget()
+			if err != nil {
+				return err
+			}
+
+			s.upstreams[i].Name = target
+		} else if s.upstreams[i].Http != nil {
+			s.upstreams[i].Name = s.upstreams[i].Http.Target
+		}
 	}
 
 	return nil
@@ -207,6 +261,50 @@ func WithHeaderProcessor(processHeader func(http.Header) []string) func(*Server)
 	}
 }
 
+func buildRequestWithNewTarget(r *http.Request, target *HttpClientConf) (*http.Request, error) {
+	u := *r.URL
+	u.Host = target.Target
+	if len(u.Scheme) == 0 {
+		u.Scheme = defaultHttpScheme
+	}
+
+	if len(target.Prefix) > 0 {
+		var err error
+		u.Path, err = url.JoinPath(target.Prefix, u.Path)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	return &http.Request{
+		Method:        r.Method,
+		URL:           &u,
+		Header:        r.Header.Clone(),
+		Proto:         r.Proto,
+		ProtoMajor:    r.ProtoMajor,
+		ProtoMinor:    r.ProtoMinor,
+		ContentLength: r.ContentLength,
+		Body:          io.NopCloser(r.Body),
+	}, nil
+}
+
+func createDescriptorSource(cli zrpc.Client, up Upstream) (grpcurl.DescriptorSource, error) {
+	var source grpcurl.DescriptorSource
+	var err error
+
+	if len(up.ProtoSets) > 0 {
+		source, err = grpcurl.DescriptorSourceFromProtoSets(up.ProtoSets...)
+		if err != nil {
+			return nil, err
+		}
+	} else {
+		client := grpcreflect.NewClientAuto(context.Background(), cli.Conn())
+		source = grpcurl.DescriptorSourceFromServer(context.Background(), client)
+	}
+
+	return source, nil
+}
+
 // withDialer sets a dialer to create a gRPC client.
 func withDialer(dialer func(conf zrpc.RpcClientConf) zrpc.Client) func(*Server) {
 	return func(s *Server) {