补充框架App启动

docs/example/main.go

@@ -1,9 +1,8 @@
 package main
 
 import (
-	"fmt"
+	"context"
 
-	"github.com/gin-gonic/gin"
 	"github.com/infraboard/mcube/ioc"
 	"github.com/infraboard/mcube/ioc/config/application"
 
@@ -22,16 +21,8 @@ func main() {
 	}
 
 	// 打印ioc当前托管的对象
-	fmt.Println(ioc.Config().List())
-	fmt.Println(ioc.Config().List())
-	fmt.Println(ioc.Api().List())
-
-	// 将Ioc所有业务模块的路由 都注册给Gin Root Router
-	app := application.App()
-	r := gin.Default()
-	ioc.LoadGinApi(app.HTTPPrefix(), r)
-
-	// 启动HTTP服务
-	// ioc中读取应用相关配置
-	r.Run(app.HTTP.Addr())
+	err = application.App().Start(context.Background())
+	if err != nil {
+		panic(err)
+	}
 }

ioc/config/application/application.go

@@ -3,10 +3,15 @@ package application
 import (
 	"context"
 	"fmt"
+	"os"
+	"os/signal"
+	"syscall"
 
 	"github.com/go-openapi/spec"
 	"github.com/infraboard/mcube/ioc"
+	"github.com/infraboard/mcube/ioc/config/logger"
 	"github.com/infraboard/mcube/tools/pretty"
+	"github.com/rs/zerolog"
 )
 
 func init() {
@@ -28,6 +33,11 @@ type Application struct {
 	GRPC           *Grpc  `json:"grpc" yaml:"grpc"  toml:"grpc"`
 
 	ioc.ObjectImpl
+
+	ch     chan os.Signal
+	log    *zerolog.Logger
+	ctx    context.Context
+	cancle context.CancelFunc
 }
 
 func (a *Application) HTTPPrefix() string {
@@ -43,6 +53,13 @@ func (a *Application) Name() string {
 }
 
 func (a *Application) Init() error {
+	// 处理信号量
+	ch := make(chan os.Signal, 1)
+	signal.Notify(ch, syscall.SIGTERM, syscall.SIGINT, syscall.SIGHUP, syscall.SIGQUIT)
+	a.ch = ch
+	a.log = logger.Sub("application")
+	a.ctx, a.cancle = context.WithCancel(context.Background())
+
 	if err := a.HTTP.Parse(); err != nil {
 		return err
 	}
@@ -69,11 +86,51 @@ func (a *Application) SwagerDocs(swo *spec.Swagger) {
 }
 
 func (a *Application) Start(ctx context.Context) error {
+	a.log.Info().Msgf("loaded configs: %s", ioc.Config().List())
+	a.log.Info().Msgf("loaded controllers: %s", ioc.Config().List())
+	a.log.Info().Msgf("loaded apis: %s", ioc.Api().List())
+
 	if *a.HTTP.Enable {
-		a.HTTP.Start(ctx)
+		go a.HTTP.Start(ctx, a.HandleError)
 	}
 	if *a.GRPC.Enable {
-		a.GRPC.Start(ctx)
+		go a.GRPC.Start(ctx, a.HandleError)
 	}
+
+	a.waitSign()
 	return nil
 }
+
+func (a *Application) HandleError(err error) {
+	if err != nil {
+		a.log.Error().Msg(err.Error())
+	}
+}
+
+func (a *Application) waitSign() {
+	defer a.cancle()
+
+	for sg := range a.ch {
+		switch v := sg.(type) {
+		default:
+			a.log.Info().Msgf("receive signal '%v', start graceful shutdown", v.String())
+
+			if *a.GRPC.Enable {
+				if err := a.GRPC.Stop(a.ctx); err != nil {
+					a.log.Error().Msgf("grpc graceful shutdown err: %s, force exit", err)
+				} else {
+					a.log.Info().Msg("grpc service stop complete")
+				}
+			}
+
+			if *a.HTTP.Enable {
+				if err := a.HTTP.Stop(a.ctx); err != nil {
+					a.log.Error().Msgf("http graceful shutdown err: %s, force exit", err)
+				} else {
+					a.log.Info().Msgf("http service stop complete")
+				}
+			}
+			return
+		}
+	}
+}

ioc/config/application/grpc.go

@@ -3,6 +3,7 @@ package application
 import (
 	"context"
 	"fmt"
+	"net"
 
 	"github.com/infraboard/mcube/grpc/middleware/recovery"
 	"github.com/infraboard/mcube/ioc"
@@ -13,17 +14,10 @@ import (
 )
 
 func NewDefaultGrpc() *Grpc {
-	rc := recovery.NewInterceptor(recovery.NewZeroLogRecoveryHandler())
-	grpcServer := grpc.NewServer(grpc.ChainUnaryInterceptor(
-		rc.UnaryServerInterceptor(),
-		otelgrpc.UnaryServerInterceptor(),
-	))
-	fmt.Println(grpcServer)
-
 	return &Grpc{
-		Host: "127.0.0.1",
-		Port: 18080,
-		log:  logger.Sub("grpc"),
+		Host:           "127.0.0.1",
+		Port:           18080,
+		EnableRecovery: true,
 	}
 }
 
@@ -37,9 +31,20 @@ type Grpc struct {
 	CertFile  string `json:"cert_file" yaml:"cert_file" toml:"cert_file" env:"GRPC_CERT_FILE"`
 	KeyFile   string `json:"key_file" yaml:"key_file" toml:"key_file" env:"GRPC_KEY_FILE"`
 
+	// 开启recovery恢复
+	EnableRecovery bool `json:"enable_recovery" yaml:"enable_recovery" toml:"enable_recovery" env:"GRPC_ENABLE_RECOVERY"`
+	// 开启Trace
+	EnableTrace bool `json:"enable_trace" yaml:"enable_trace" toml:"enable_trace" env:"GRPC_ENABLE_TRACE"`
+
 	// 解析后的数据
-	svr *grpc.Server
-	log *zerolog.Logger
+	interceptors []grpc.UnaryServerInterceptor
+	svr          *grpc.Server
+	log          *zerolog.Logger
+
+	// 启动后执行
+	PostStart func(context.Context) error
+	// 关闭前执行
+	PreStop func(context.Context) error
 }
 
 func (g *Grpc) setEnable(v bool) {
@@ -51,16 +56,68 @@ func (g *Grpc) Addr() string {
 }
 
 func (g *Grpc) Parse() error {
+	g.log = logger.Sub("grpc")
+
 	if g.Enable == nil {
 		g.setEnable(ioc.GrpcControllerCount() > 0)
 	}
 	return nil
 }
 
-func (g *Grpc) Start(ctx context.Context) error {
-	return nil
+func (g *Grpc) AddInterceptors(interceptors ...grpc.UnaryServerInterceptor) {
+	g.interceptors = append(g.interceptors, interceptors...)
+}
+
+func (g *Grpc) Interceptors() (interceptors []grpc.UnaryServerInterceptor) {
+	if g.EnableRecovery {
+		interceptors = append(interceptors,
+			recovery.NewInterceptor(recovery.NewZeroLogRecoveryHandler()).
+				UnaryServerInterceptor())
+	}
+	if g.EnableTrace {
+		interceptors = append(interceptors, otelgrpc.UnaryServerInterceptor())
+	}
+
+	interceptors = append(interceptors, g.interceptors...)
+	return
+}
+
+type ServiceInfoCtxKey struct{}
+
+func (g *Grpc) Start(ctx context.Context, cb ErrHandler) {
+	g.svr = grpc.NewServer(grpc.ChainUnaryInterceptor(g.Interceptors()...))
+
+	// 装载所有GRPC服务
+	ioc.LoadGrpcController(g.svr)
+
+	// 启动GRPC服务
+	lis, err := net.Listen("tcp", g.Addr())
+	if err != nil {
+		cb(fmt.Errorf("listen grpc tcp conn error, %s", err))
+		return
+	}
+
+	// 启动后勾子
+	ctx = context.WithValue(ctx, ServiceInfoCtxKey{}, g.svr.GetServiceInfo())
+	if g.PostStart != nil {
+		if err := g.PostStart(ctx); err != nil {
+			cb(err)
+			return
+		}
+	}
+
+	g.log.Info().Msgf("GRPC 服务监听地址: %s", g.Addr())
+	cb(g.svr.Serve(lis))
 }
 
 func (g *Grpc) Stop(ctx context.Context) error {
+	// 停止之前的Hook
+	if g.PreStop != nil {
+		if err := g.PreStop(ctx); err != nil {
+			return err
+		}
+	}
+
+	g.svr.GracefulStop()
 	return nil
 }

ioc/config/application/http.go

@@ -21,7 +21,6 @@ func NewDefaultHttp() *Http {
 		WriteTimeoutSecond:      60,
 		IdleTimeoutSecond:       300,
 		MaxHeaderSize:           "16kb",
-		log:                     logger.Sub("http"),
 		WEB_FRAMEWORK:           WEB_FRAMEWORK_GO_RESTFUL,
 		routerBuilders: map[WEB_FRAMEWORK]RouterBuilder{
 			WEB_FRAMEWORK_GO_RESTFUL: NewGoRestfulRouterBuilder(),
@@ -102,6 +101,8 @@ func (h *Http) setEnable(v bool) {
 
 // 配置数据解析
 func (h *Http) Parse() error {
+	h.log = logger.Sub("http")
+
 	if h.Enable == nil {
 		h.setEnable(ioc.Api().Count() > 0)
 	}
@@ -151,29 +152,26 @@ func (h *Http) BuildRouter() error {
 	return nil
 }
 
+type ErrHandler func(error)
+
 // Start 启动服务
-func (h *Http) Start(ctx context.Context) error {
+func (h *Http) Start(ctx context.Context, cb ErrHandler) {
 	if err := h.BuildRouter(); err != nil {
-		return fmt.Errorf("build router error, %s", err)
+		cb(fmt.Errorf("build http router error, %s", err))
+		return
 	}
 
 	// 启动 HTTP服务
 	h.log.Info().Msgf("HTTP服务启动成功, 监听地址: %s", h.Addr())
-	if err := h.server.ListenAndServe(); err != nil {
-		if err != http.ErrServerClosed {
-			return fmt.Errorf("start service error, %s", err.Error())
-		}
-	}
-
-	return nil
+	cb(h.server.ListenAndServe())
 }
 
 // Stop 停止server
 func (h *Http) Stop(ctx context.Context) error {
 	h.log.Info().Msg("start graceful shutdown")
 	// 优雅关闭HTTP服务
 	if err := h.server.Shutdown(ctx); err != nil {
-		return fmt.Errorf("graceful shutdown timeout, force exit")
+		return fmt.Errorf("http graceful shutdown timeout, force exit")
 	}
 	return nil
 }

ioc/config/application/http_gorestful.go

@@ -8,27 +8,26 @@ import (
 	"github.com/infraboard/mcube/ioc/apps/apidoc"
 	"github.com/infraboard/mcube/ioc/apps/health"
 	"github.com/infraboard/mcube/ioc/config/logger"
-	"github.com/rs/zerolog"
 	"go.opentelemetry.io/contrib/instrumentation/github.com/emicklei/go-restful/otelrestful"
 )
 
 func NewGoRestfulRouterBuilder() *GoRestfulRouterBuilder {
 	return &GoRestfulRouterBuilder{
-		log:  logger.Sub("http"),
 		conf: &BuildConfig{},
 	}
 }
 
 type GoRestfulRouterBuilder struct {
 	conf *BuildConfig
-	log  *zerolog.Logger
 }
 
 func (b *GoRestfulRouterBuilder) Config(c *BuildConfig) {
 	b.conf = c
 }
 
 func (b *GoRestfulRouterBuilder) Build() (http.Handler, error) {
+	log := logger.Sub("go-restful")
+
 	r := restful.DefaultContainer
 	restful.DefaultResponseContentType(restful.MIME_JSON)
 	restful.DefaultRequestContentType(restful.MIME_JSON)
@@ -65,14 +64,14 @@ func (b *GoRestfulRouterBuilder) Build() (http.Handler, error) {
 	// API Doc
 	if App().HTTP.EnableApiDoc {
 		r.Add(apidoc.APIDocs(App().HTTP.ApiDocPath, App().SwagerDocs))
-		b.log.Info().Msgf("Get the API Doc using http://%s%s", App().HTTP.Addr(), App().HTTP.ApiDocPath)
+		log.Info().Msgf("Get the API Doc using http://%s%s", App().HTTP.Addr(), App().HTTP.ApiDocPath)
 	}
 
 	// HealthCheck
 	if App().HTTP.EnableHealthCheck {
 		hc := health.NewDefaultHealthChecker()
 		r.Add(hc.WebService())
-		b.log.Info().Msgf("健康检查地址: http://%s%s", App().HTTP.Addr(), hc.HealthCheckPath)
+		log.Info().Msgf("健康检查地址: http://%s%s", App().HTTP.Addr(), hc.HealthCheckPath)
 	}
 
 	return r, nil