package micro import ( "context" "crypto/md5" "encoding/hex" "git.nspix.com/golang/micro/gateway" "git.nspix.com/golang/micro/gateway/http" "git.nspix.com/golang/micro/log" "git.nspix.com/golang/micro/registry" "git.nspix.com/golang/micro/utils/docker" "git.nspix.com/golang/micro/utils/net/ip" "git.nspix.com/golang/micro/utils/unsafestr" "net" "os" "os/signal" "strings" "sync" "syscall" "time" ) type Service struct { opts *Options ctx context.Context cancelFunc context.CancelFunc registry registry.Registry node *registry.ServiceNode listener net.Listener gateway *gateway.Gateway wg sync.WaitGroup httpSvr *http.Server upTime time.Time } func (svr *Service) wrapSync(f func()) { svr.wg.Add(1) go func() { f() svr.wg.Done() }() } func (svr *Service) eventLoop() { var ( err error ticker *time.Ticker ) ticker = time.NewTicker(time.Second * 10) defer ticker.Stop() for { select { case <-ticker.C: if err = svr.registry.Register(svr.node); err != nil { log.Warnf("registry service %s error: %s", svr.opts.Name, err.Error()) } case <-svr.ctx.Done(): return } } } func (svr *Service) Handle(method string, cb HandleFunc) { if svr.opts.EnableHttp { //convert url path := strings.ReplaceAll(method, ".", "/") if path[0] != '/' { path = "/" + path } svr.httpSvr.Handle("POST", path, func(ctx *http.Context) (err error) { return cb(ctx) }) } //启动RPC功能 if svr.opts.EnableRPC { } return } func (svr *Service) generateInstance() { var ( err error id string dockerID string tcpAddr *net.TCPAddr ) if id, err = docker.SelfContainerID(); err != nil { //生成唯一ID e5 := md5.New() e5.Write(unsafestr.StringToBytes(svr.opts.Name)) e5.Write(unsafestr.StringToBytes(svr.opts.Version)) id = hex.EncodeToString(e5.Sum(nil)) } else { dockerID = id } svr.node = ®istry.ServiceNode{ ID: id, Name: svr.opts.Name, Version: svr.opts.Version, Metadata: map[string]string{}, Addresses: make(map[string]registry.Addr), } ipLocal := ip.InternalIP() svr.node.Address = ipLocal if tcpAddr, err = net.ResolveTCPAddr("tcp", svr.listener.Addr().String()); err == nil { svr.node.Port = tcpAddr.Port } svr.node.Metadata["docker-id"] = dockerID if svr.opts.EnableHttp { svr.node.Metadata["enable-http"] = "true" } if svr.opts.EnableRPC { svr.node.Metadata["enable-rpc"] = "true" } } func (svr *Service) prepare() (err error) { log.Prefix(svr.opts.Name) if svr.listener, err = net.ListenTCP("tcp", nil); err != nil { return } log.Infof("listener on: %s", svr.listener.Addr()) svr.gateway = gateway.New(svr.listener) svr.generateInstance() svr.wrapSync(func() { svr.gateway.Run(svr.ctx) }) svr.wrapSync(func() { svr.eventLoop() }) if svr.opts.EnableHttp { l := gateway.NewListener(svr.listener.Addr()) if err = svr.gateway.Attaches([][]byte{[]byte("GET"), []byte("POST"), []byte("PUT"), []byte("DELETE"), []byte("OPTIONS")}, l); err == nil { svr.wrapSync(func() { _ = svr.httpSvr.Serve(l) }) svr.httpSvr.Handle("GET", "/healthy", func(ctx *http.Context) (err error) { return ctx.Success(map[string]interface{}{ "healthy": "healthy", "uptime": time.Now().Sub(svr.upTime).String(), }) }) log.Infof("attach http server success") } else { log.Warnf("attach http listener error: %s", err.Error()) } } if svr.opts.EnableHttp { l := gateway.NewListener(svr.listener.Addr()) if err = svr.gateway.Attach([]byte("RPC"), l); err == nil { serve := http.New() svr.wrapSync(func() { _ = serve.Serve(l) }) } } if err = svr.registry.Register(svr.node); err == nil { log.Infof("register service %s successful", svr.opts.Name) } return } func (svr *Service) destroy() (err error) { log.Infof("stopping") svr.cancelFunc() if err = svr.registry.Deregister(svr.node); err != nil { log.Warnf("deregister service %s error: %s", svr.opts.Name, err.Error()) } else { log.Infof("deregister service %s successful", svr.opts.Name) } if svr.listener != nil { if err = svr.listener.Close(); err != nil { log.Warnf(err.Error()) } } log.Infof("stopped") return } func (svr *Service) Run() (err error) { if err = svr.prepare(); err != nil { return } //start server if svr.opts.Server != nil { if err = svr.opts.Server.Start(svr.ctx); err != nil { return } } //waiting ch := make(chan os.Signal, 1) signal.Notify(ch, syscall.SIGTERM, syscall.SIGINT, syscall.SIGQUIT, syscall.SIGKILL) select { case <-ch: case <-svr.ctx.Done(): } //stop server if svr.opts.Server != nil { err = svr.opts.Server.Stop() } return svr.destroy() } func New(opts ...OptionMiddleware) *Service { o := NewOptions() for _, opt := range opts { opt(o) } svr := &Service{ opts: o, upTime: time.Now(), httpSvr: http.New(), registry: registry.DefaultRegistry, } svr.ctx, svr.cancelFunc = context.WithCancel(o.Context) return svr }