You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
137 lines
2.8 KiB
137 lines
2.8 KiB
package kratos
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"os"
|
|
"os/signal"
|
|
"sync"
|
|
"syscall"
|
|
|
|
"github.com/go-kratos/kratos/v2/log"
|
|
"github.com/go-kratos/kratos/v2/registry"
|
|
"github.com/go-kratos/kratos/v2/transport"
|
|
|
|
"github.com/google/uuid"
|
|
"golang.org/x/sync/errgroup"
|
|
)
|
|
|
|
// App is an application components lifecycle manager
|
|
type App struct {
|
|
opts options
|
|
ctx context.Context
|
|
cancel func()
|
|
instance *registry.ServiceInstance
|
|
log *log.Helper
|
|
}
|
|
|
|
// New create an application lifecycle manager.
|
|
func New(opts ...Option) *App {
|
|
options := options{
|
|
ctx: context.Background(),
|
|
logger: log.DefaultLogger,
|
|
sigs: []os.Signal{syscall.SIGTERM, syscall.SIGQUIT, syscall.SIGINT},
|
|
}
|
|
if id, err := uuid.NewUUID(); err == nil {
|
|
options.id = id.String()
|
|
}
|
|
for _, o := range opts {
|
|
o(&options)
|
|
}
|
|
ctx, cancel := context.WithCancel(options.ctx)
|
|
return &App{
|
|
ctx: ctx,
|
|
cancel: cancel,
|
|
opts: options,
|
|
log: log.NewHelper(options.logger),
|
|
}
|
|
}
|
|
|
|
// Run executes all OnStart hooks registered with the application's Lifecycle.
|
|
func (a *App) Run() error {
|
|
a.log.Infow(
|
|
"service_id", a.opts.id,
|
|
"service_name", a.opts.name,
|
|
"service_version", a.opts.version,
|
|
)
|
|
instance, err := a.buildInstance()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
ctx := NewContext(a.ctx, AppInfo{
|
|
ID: a.opts.id,
|
|
Name: a.opts.name,
|
|
Version: a.opts.version,
|
|
})
|
|
eg, ctx := errgroup.WithContext(ctx)
|
|
wg := sync.WaitGroup{}
|
|
for _, srv := range a.opts.servers {
|
|
srv := srv
|
|
eg.Go(func() error {
|
|
<-ctx.Done() // wait for stop signal
|
|
return srv.Stop(ctx)
|
|
})
|
|
wg.Add(1)
|
|
eg.Go(func() error {
|
|
wg.Done()
|
|
return srv.Start(ctx)
|
|
})
|
|
}
|
|
wg.Wait()
|
|
if a.opts.registrar != nil {
|
|
if err := a.opts.registrar.Register(a.opts.ctx, instance); err != nil {
|
|
return err
|
|
}
|
|
a.instance = instance
|
|
}
|
|
c := make(chan os.Signal, 1)
|
|
signal.Notify(c, a.opts.sigs...)
|
|
eg.Go(func() error {
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
case <-c:
|
|
a.Stop()
|
|
}
|
|
}
|
|
})
|
|
if err := eg.Wait(); err != nil && !errors.Is(err, context.Canceled) {
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Stop gracefully stops the application.
|
|
func (a *App) Stop() error {
|
|
if a.opts.registrar != nil && a.instance != nil {
|
|
if err := a.opts.registrar.Deregister(a.opts.ctx, a.instance); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
if a.cancel != nil {
|
|
a.cancel()
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (a *App) buildInstance() (*registry.ServiceInstance, error) {
|
|
if len(a.opts.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
|
|
}
|
|
a.opts.endpoints = append(a.opts.endpoints, e)
|
|
}
|
|
}
|
|
}
|
|
return ®istry.ServiceInstance{
|
|
ID: a.opts.id,
|
|
Name: a.opts.name,
|
|
Version: a.opts.version,
|
|
Metadata: a.opts.metadata,
|
|
Endpoints: a.opts.endpoints,
|
|
}, nil
|
|
}
|
|
|