1 Star 0 Fork 0

s-dy / yogurt

加入 Gitee
与超过 1200万 开发者一起发现、参与优秀开源项目,私有仓库也完全免费 :)
免费加入
克隆/下载
app.go 4.04 KB
一键复制 编辑 原始数据 按行查看 历史
s-dy 提交于 2022-10-14 18:30 . log, errors, app
package yogurt
import (
"context"
"errors"
"gitee.com/sdynasty/yogurt/log"
"gitee.com/sdynasty/yogurt/registry"
"gitee.com/sdynasty/yogurt/transport"
"github.com/google/uuid"
"golang.org/x/sync/errgroup"
"os"
"os/signal"
"sync"
"syscall"
"time"
)
type AppInfo interface {
ID() string
Name() string
Version() string
Metadata() map[string]string
Endpoint() []string
}
type App struct {
opts options
ctx context.Context
cancel func()
mu sync.Mutex
instance *registry.ServiceInstance // App.Run的时候根据options.servers build
}
func New(opts ...Option) *App {
o := options{
ctx: context.Background(),
sigs: []os.Signal{syscall.SIGTERM},
registrarTimeout: 10 * time.Second,
stopTimeout: 10 * time.Second,
}
if id, err := uuid.NewUUID(); err == nil {
o.id = id.String()
}
for _, opt := range opts {
opt(&o)
}
if o.logger != nil {
log.SetLogger(o.logger)
}
ctx, cancel := context.WithCancel(o.ctx)
return &App{
ctx: ctx,
cancel: cancel,
opts: o,
}
}
// ID returns app instance id.
func (a *App) ID() string { return a.opts.id }
// Name returns service name.
func (a *App) Name() string { return a.opts.name }
// Version returns app version.
func (a *App) Version() string { return a.opts.version }
// Metadata returns service metadata.
func (a *App) Metadata() map[string]string { return a.opts.metadata }
func (a *App) Endpoint() []string {
if a.instance != nil {
return a.instance.Endpoints
}
return nil
}
func (a *App) Run() error {
instance, err := a.buildInstance()
if err != nil {
return nil
}
a.mu.Lock()
a.instance = instance
a.mu.Unlock()
eg, ctx := errgroup.WithContext(NewContext(a.ctx, a))
wg := sync.WaitGroup{} // 全部启动后再注册
for _, srv := range a.opts.servers {
srv := srv
eg.Go(func() error {
<-ctx.Done() // wait for stop signal,优雅关闭
stopCtx, cancel := context.WithTimeout(NewContext(a.opts.ctx, a), a.opts.stopTimeout)
defer cancel()
return srv.Stop(stopCtx)
})
wg.Add(1)
eg.Go(func() error {
wg.Done() // here is to ensure server start has begun running before register, so defer is not needed
return srv.Start(NewContext(a.opts.ctx, a))
})
}
wg.Wait()
if a.opts.registrar != nil {
rctx, rcancel := context.WithTimeout(ctx, a.opts.registrarTimeout)
defer rcancel()
if err := a.opts.registrar.Register(rctx, instance); err != nil {
return err
}
}
c := make(chan os.Signal, 1)
signal.Notify(c, a.opts.sigs...)
eg.Go(func() error {
// 阻塞eg
select {
case <-ctx.Done():
return nil
case <-c:
return a.Stop()
}
})
if err := eg.Wait(); err != nil && !errors.Is(err, context.Canceled) { //App.Stop()主动退出,不需要报错(父cancel(),子Done;子cancel(),父不Done)
return err
}
return nil
}
func (a *App) Stop() error {
a.mu.Lock()
instance := a.instance
a.mu.Unlock()
if a.opts.registrar != nil && instance != nil {
ctx, cancel := context.WithTimeout(NewContext(a.ctx, a), a.opts.registrarTimeout)
defer cancel()
a.opts.registrar.Deregister(ctx, instance)
}
if a.cancel != nil {
a.cancel()
}
return nil
}
// 主要是根据server生成endpoints
func (a *App) buildInstance() (*registry.ServiceInstance, error) {
endpoints := make([]string, 0, len(a.opts.endpoints))
for _, e := range a.opts.endpoints {
endpoints = append(endpoints, e.String())
}
// 优先取endpoints,没有再从server中取
if len(endpoints) == 0 {
for _, srv := range a.opts.servers {
if r, ok := srv.(transport.Endpointer); ok {
e, err := r.Endpoint()
if err != nil {
return nil, err
}
endpoints = append(endpoints, e.String())
}
}
}
return &registry.ServiceInstance{
ID: a.opts.id,
Name: a.opts.name,
Version: a.opts.version,
Metadata: a.opts.metadata,
Endpoints: endpoints,
}, nil
}
type appKey struct{}
func NewContext(ctx context.Context, s AppInfo) context.Context {
return context.WithValue(ctx, appKey{}, s)
}
func FromContext(ctx context.Context) (s AppInfo, ok bool) {
s, ok = ctx.Value(appKey{}).(AppInfo)
return
}
1
https://gitee.com/sdynasty/yogurt.git
git@gitee.com:sdynasty/yogurt.git
sdynasty
yogurt
yogurt
01d4da0b550e

搜索帮助