package nnet import ( "git.noahlan.cn/noahlan/nnet/config" "git.noahlan.cn/noahlan/nnet/connection" "git.noahlan.cn/noahlan/nnet/lifetime" "git.noahlan.cn/noahlan/nnet/packet" rt "git.noahlan.cn/noahlan/nnet/router" "git.noahlan.cn/noahlan/nnet/scheduler" "git.noahlan.cn/noahlan/nnet/serialize" "git.noahlan.cn/noahlan/nnet/session" "git.noahlan.cn/noahlan/ntools-go/core/nlog" "github.com/panjf2000/ants/v2" "math" "net" ) // Engine 引擎 type Engine struct { config.EngineConf // 引擎配置 middlewares []rt.Middleware // 中间件 routes []rt.Route // 路由 router rt.Router // 消息处理器 dieChan chan struct{} // 应用程序退出信号 pipeline connection.Pipeline // 消息管道 packerBuilder packet.PackerBuilder // 封包、拆包器 serializer serialize.Serializer // 消息 序列化/反序列化 goPool *ants.Pool // goroutine池 connManager *connection.Manager // 连接管理器 lifetime *lifetime.Mgr // 生命周期 sessIdMgr *session.IDMgr // SessionId管理器 } func NewEngine(conf config.EngineConf, opts ...RunOption) *Engine { ngin := &Engine{ EngineConf: conf, middlewares: make([]rt.Middleware, 0), routes: make([]rt.Route, 0), router: rt.NewDefaultRouter(), packerBuilder: nil, serializer: nil, dieChan: make(chan struct{}), pipeline: connection.NewPipeline(), connManager: connection.NewManager(), lifetime: lifetime.NewLifetime(), sessIdMgr: session.NewSessionIDMgr(), goPool: nil, } for _, opt := range opts { opt(ngin) } if ngin.goPool == nil { ngin.goPool, _ = ants.NewPool(math.MaxInt32) } return ngin } func (ngin *Engine) Use(middleware ...rt.Middleware) { ngin.middlewares = append(ngin.middlewares, middleware...) } func (ngin *Engine) AddRoutes(rs ...rt.Route) { ngin.routes = append(ngin.routes, rs...) err := ngin.bindRoutes() nlog.Must(err) } func (ngin *Engine) bindRoutes() error { for _, fr := range ngin.routes { if err := ngin.bindRoute(fr); err != nil { return err } } return nil } func (ngin *Engine) bindRoute(route rt.Route) error { // TODO default middleware chain := rt.NewChain() // build chain for _, middleware := range ngin.middlewares { chain.Append(rt.ConvertMiddleware(middleware)) } return ngin.router.Register(route.Matches, route.Handler) } func (ngin *Engine) setup() error { if err := ngin.bindRoutes(); err != nil { return err } if err := ngin.goPool.Submit(func() { scheduler.Schedule(ngin.TaskTimerPrecision) }); err != nil { return err } return nil } func (ngin *Engine) Stop() { nlog.Infof("%s is stopping...", ngin.LogPrefix()) close(ngin.dieChan) scheduler.Close() } func (ngin *Engine) handle(conn net.Conn) *connection.Connection { nc := connection.NewConnection( ngin.sessIdMgr.SessionID(), conn, connection.Config{LogDebug: ngin.ShallLogDebug(), LogPrefix: ngin.LogPrefix()}, ngin.packerBuilder, ngin.serializer, ngin.pipeline, ngin.router.Handle, ) nc.Serve() err := ngin.connManager.Store(connection.DefaultGroupName, nc) nlog.Must(err) // dieChan go func() { // lifetime ngin.lifetime.Open(nc) select { case <-nc.DieChan(): scheduler.PushTask(func() { ngin.lifetime.Close(nc) }) _ = ngin.connManager.Remove(nc) } }() return nc }