package cmd import ( "context" "fmt" "git.5iot.com/iotlcm-public/iotlcm-upstream-chan/internal/controller/cardquery" "git.5iot.com/iotlcm-public/iotlcm-upstream-chan/internal/controller/infopb" "git.5iot.com/iotlcm-public/iotlcm-upstream-chan/internal/controller/msgcallpb" "git.5iot.com/iotlcm-public/iotlcm-upstream-chan/internal/model" "git.5iot.com/iotlcm-public/iotlcm-upstream-chan/internal/service" sharedtoken "git.5iot.com/iotlcm-public/iotlcm-upstream-chan/internal/token" "git.5iot.com/iotlcm-public/iotlcm-upstream-chan/internal/pub/xjpubconsts" "git.5iot.com/iotlcm-public/iotlcm-upstream-chan/internal/pub/xjpubservice" "github.com/gogf/gf/contrib/rpc/grpcx/v2" "github.com/gogf/gf/v2/frame/g" "github.com/gogf/gf/v2/net/ghttp" "github.com/gogf/gf/v2/net/gsvc" "github.com/gogf/gf/v2/os/gcmd" "github.com/gogf/gf/v2/util/gconv" "google.golang.org/grpc" "google.golang.org/grpc/reflection" ) const ( ServiceNamePrefix = "iotlcmupstreamchan" ) var ( gGrpcPort = 8700 gAccessIP = "0.0.0.0" ) var ( Main = gcmd.Command{ Name: "main", Usage: "main", Brief: "start http server", Func: func(ctx context.Context, parser *gcmd.Parser) (err error) { s := g.Server() gGrpcPort = gconv.Int(gcmd.GetOptWithEnv("grpcport")) gAccessIP = gconv.String(gcmd.GetOptWithEnv("accessip")) if len(gAccessIP) == 0 { gAccessIP = "0.0.0.0" } if gGrpcPort == 0 { gGrpcPort = 8110 } nodeName := fmt.Sprintf("%s", ServiceNamePrefix) xjpubservice.Info().SetInfoName(xjpubconsts.ServiceTypeAccess, fmt.Sprintf("%s-http-svc", ServiceNamePrefix), "上游运营商渠道服务", "上游运营商渠道服务") xjpubservice.Info().AddInfoData("grpcport", gGrpcPort) xjpubservice.Info().AddInfoData("accessip", gAccessIP) xjpubservice.Info().AddInfoData("nodeName", nodeName) xjpubservice.InfoRpc().SetInfoName(xjpubconsts.ServiceTypeAccess, fmt.Sprintf("%s-grpc-svc", ServiceNamePrefix), "上游运营商渠道服务", "上游运营商渠道服务") xjpubservice.InfoRpc().AddInfoData("grpcport", gGrpcPort) xjpubservice.InfoRpc().AddInfoData("accessip", gAccessIP) xjpubservice.InfoRpc().AddInfoData("nodeName", nodeName) bindHTTPRoutes(s) err = service.ChannelConfig().Start(ctx) if err != nil { return err } err = service.UpstreamAPI().Start(ctx) if err != nil { return err } go func() { c := grpcx.Server.NewConfig() c.Endpoints = []string{fmt.Sprintf("%s:%d", gAccessIP, gGrpcPort)} c.Options = append(c.Options, []grpc.ServerOption{ grpcx.Server.ChainUnary( grpcx.Server.UnaryValidate, ), }...) sGRPC := grpcx.Server.New(c) sGRPC.Service(&gsvc.LocalService{ Name: fmt.Sprintf("%s-grpc-svc", ServiceNamePrefix), Endpoints: gsvc.Endpoints{gsvc.NewEndpoint(fmt.Sprintf("%s:%d", gAccessIP, gGrpcPort))}, Metadata: gsvc.Metadata{ gsvc.MDProtocol: "grpc", gsvc.MDInsecure: true, }, }) infopb.Register(sGRPC) msgcallpb.Register(sGRPC) reflection.Register(sGRPC.Server) g.Log().Infof(ctx, "GRPC Server启动, listenPort:%d", gGrpcPort) sGRPC.Run() g.Log().Infof(ctx, "GRPC Server堵塞运行结束") }() s.Run() service.UpstreamAPI().Stop(ctx) service.ChannelConfig().Stop(ctx) return nil }, } ) type sharedTokenGetter interface { GetSharedToken(ctx context.Context) (sharedtoken.OneLinkToken, error) } func bindHTTPRoutes(s *ghttp.Server) { s.Group("/", func(group *ghttp.RouterGroup) { group.Middleware(ghttp.MiddlewareHandlerResponse) group.Bind( cardquery.NewV1(), ) // Internal control endpoint used to apply a single channel change // without waiting for the periodic full reload. group.POST("/internal/channel/test", func(r *ghttp.Request) { input := new(model.ChannelTestInput) if err := r.Parse(input); err != nil { r.SetError(err) return } output, err := service.UpstreamAPI().TestChannel(r.Context(), input) if err != nil { r.SetError(err) return } r.Response.WriteJson(g.Map{"code": 0, "message": "success", "data": output}) }) group.POST("/internal/channel/refresh", func(r *ghttp.Request) { chanKey := r.GetQuery("chanKey").String() if chanKey == "" { chanKey = r.Get("chanKey").String() } if err := service.UpstreamAPI().RefreshChannel(r.Context(), chanKey); err != nil { r.SetError(err) return } r.Response.WriteJson(g.Map{ "code": 0, "message": "success", "data": g.Map{"chanKey": chanKey, "refreshed": true}, }) }) group.POST("/internal/channel/remove", func(r *ghttp.Request) { chanKey := r.GetQuery("chanKey").String() if chanKey == "" { chanKey = r.Get("chanKey").String() } if err := service.UpstreamAPI().RemoveChannel(r.Context(), chanKey); err != nil { r.SetError(err) return } r.Response.WriteJson(g.Map{ "code": 0, "message": "success", "data": g.Map{"chanKey": chanKey, "removed": true}, }) }) group.POST("/internal/channel/shared-token", func(r *ghttp.Request) { chanKey := r.Get("chanKey").String() if chanKey == "" { r.Response.WriteJson(g.Map{ "code": 50, "message": "chanKey不能为空", }) return } channelCache, ok := service.UpstreamAPI().GetChan(chanKey) if !ok || channelCache.Instance == nil { if err := service.UpstreamAPI().RefreshChannel(r.Context(), chanKey); err != nil { g.Log().Errorf(r.Context(), "shared-token refresh channel failed, chanKey: %s, err: %+v", chanKey, err) r.Response.WriteJson(g.Map{ "code": 50, "message": "通道实例化失败", }) return } channelCache, ok = service.UpstreamAPI().GetChan(chanKey) if !ok || channelCache.Instance == nil { r.Response.WriteJson(g.Map{ "code": 50, "message": "通道不存在", }) return } } provider, ok := channelCache.Instance.(sharedTokenGetter) if !ok { r.Response.WriteJson(g.Map{ "code": 50, "message": "该通道不支持共享Token", }) return } tok, err := provider.GetSharedToken(r.Context()) if err != nil { g.Log().Errorf(r.Context(), "shared-token get token failed, chanKey: %s, err: %+v", chanKey, err) r.Response.WriteJson(g.Map{ "code": 50, "message": "获取共享Token失败", }) return } if tok.Value == "" { r.Response.WriteJson(g.Map{ "code": 50, "message": "该通道尚未获取共享Token", }) return } chanAlias := "" if channelConf, ok := service.ChannelConfig().GetChannel(chanKey); ok && channelConf.Channel != nil { chanAlias = channelConf.Channel.ChanAlias } tokenExpire := "" if !tok.ExpiresAt.IsZero() { // Go time layouts use the reference date, not PHP-style format tokens. tokenExpire = tok.ExpiresAt.Format("2006-01-02 15:04:05") } r.Response.WriteJson(g.Map{ "code": 0, "message": "success", "data": g.Map{ "chanAlias": chanAlias, "chanKey": chanKey, "accessToken": tok.Value, "accessTokenExpire": tokenExpire, }, }) }) }) }