package gateway import ( "context" "encoding/json" "net/http" "net/url" "sig-pub/api/pb" "sig-pub/pkg/grpc/generic" "sig-pub/pkg/grpc/session" "sig-pub/pkg/resp" "sig-pub/pkg/utils/strs" "sig-pub/pkg/zlog" "strings" "time" "github.com/bytedance/sonic" "github.com/fasthttp/router" "github.com/savsgio/gotils/strconv" "github.com/valyala/fasthttp" "google.golang.org/grpc" "google.golang.org/protobuf/encoding/protojson" ) const ( ReadTimeout = 20 * time.Second WriteTimeout = 20 * time.Second IdleTimeout = 120 * time.Second ) type FastGatewayServer struct { sigServerURL *url.URL grpcGenericClientFactory *generic.GrpcGenericClientFactory svr *fasthttp.Server sigClient *fasthttp.HostClient } func NewFastGatewayServer( sigServerURL *url.URL, grpcGenericClientFactory *generic.GrpcGenericClientFactory, ) *FastGatewayServer { return &FastGatewayServer{ sigServerURL: sigServerURL, grpcGenericClientFactory: grpcGenericClientFactory, } } func (g *FastGatewayServer) Init() (err error) { // fasthttp router r := router.New() r.ANY("/api/sig/{path:*}", g.reverseProxyHandler) r.POST("/api/v1/{path:*}", g.reverseProxyGrpcGenericCall) g.svr = &fasthttp.Server{ Handler: r.Handler, ReadTimeout: ReadTimeout, WriteTimeout: WriteTimeout, IdleTimeout: IdleTimeout, MaxConnsPerIP: 1024 * 4, MaxRequestsPerConn: 0, // 0 = unlimited ReduceMemoryUsage: true, } // sig http service client g.sigClient = &fasthttp.HostClient{ Addr: g.sigServerURL.Host, Name: "gateway", MaxConns: 512, // 根据实际业务调 ReadTimeout: ReadTimeout, WriteTimeout: WriteTimeout, MaxConnDuration: 10 * time.Minute, DisableHeaderNamesNormalizing: true, } return } func (g *FastGatewayServer) Run(addr string) (err error) { return g.svr.ListenAndServe(addr) } func (g *FastGatewayServer) reverseProxyHandler(ctx *fasthttp.RequestCtx) { path := ctx.UserValue("path") zlog.Infof("path: %s", path) req := &ctx.Request resp := &ctx.Response req.Header.Set("Connection", "keep-alive") err := g.sigClient.Do(req, resp) if err != nil { zlog.Errorf("reverse proxy error: url=%s, %v", strconv.B2S(ctx.RequestURI()), err) } } func responseJSON(ctx *fasthttp.RequestCtx, code int, data any) { body, err := sonic.Marshal(data) if err != nil { zlog.Errorf("marshal resp data error: url=%s, %v", strconv.B2S(ctx.RequestURI()), err) ctx.SetStatusCode(http.StatusInternalServerError) return } ctx.SetContentType("application/json; charset=utf-8") ctx.SetStatusCode(code) ctx.SetBody(body) } // reverseProxyGrpcGenericCall 代理各个grpc服务请求 func (s *FastGatewayServer) reverseProxyGrpcGenericCall(c *fasthttp.RequestCtx) { path := c.UserValue("path").(string) paths := strings.Split(path, "/") if len(paths) != 2 { responseJSON(c, http.StatusBadRequest, resp.Error("service not specified")) return } svr := strs.UpperInitialLetter(paths[0]) method := strs.UpperInitialLetter(paths[1]) if svr == "" || method == "" { responseJSON(c, http.StatusBadRequest, resp.Error("service not found")) return } if !strings.HasSuffix(svr, "Service") { svr += "Service" } // todo service white list // get request body jsonBody := c.Request.Body() ctx1, cancel := context.WithTimeout(context.Background(), time.Second*5) defer cancel() grpcGenericClient, err := s.grpcGenericClientFactory.GetClient(ctx1, svr) if err != nil { zlog.Error(err) responseJSON(c, http.StatusForbidden, resp.Error(err.Error())) return } // put session ctx := context.Background() ctx = session.PutSubject(ctx, session.NewRpcSubject("123456")) ctx, cancel = context.WithTimeout(ctx, time.Second*60) defer cancel() // todo config call options var opts []grpc.CallOption if svr == "ExchangeService" && method == "HistoryKline" { opts = append(opts, grpc.UseCompressor("snappy")) } // generic call with json rsp, err := grpcGenericClient.InvokeUnaryJsonBytes(ctx, method, jsonBody, opts...) if err != nil { if err == generic.ErrorMethodNotExists { responseJSON(c, http.StatusNotFound, resp.Error(err.Error())) return } responseJSON(c, http.StatusInternalServerError, resp.Error(err.Error())) return } var r any switch rsp := rsp.(type) { case *pb.RspIndicatorSummary: r = resp.H{ "indicator": rsp.Indicator, "summary": json.RawMessage(rsp.Summary), } default: // marshal json response bytes, err := protojson.MarshalOptions{ UseProtoNames: false, // false:lowerCamelCase, true:snake_case EmitUnpopulated: false, // 是否包含默认值 }.Marshal(rsp) if err != nil { responseJSON(c, http.StatusInternalServerError, resp.Error(err.Error())) return } r = json.RawMessage(bytes) } responseJSON(c, http.StatusOK, resp.Success(r)) }