package exegroup import ( "context" "fmt" "time" "git.blauwelle.com/go/crate/log" ) // Actor 是 [Group] 调度的执行单元; // startCtx, startCancel 需要在构造 Actor 时初始化, 在后续 start / wait 初始化会导致 data race. type Actor struct { startFunc func(ctx context.Context) error group *Group closeAfterStop chan struct{} closeAfterTimeout chan struct{} startCtx context.Context //nolint:containedctx startCancel func() name string stopTimeout time.Duration // <0表示没有超时时间 isFastStop bool } func (actor *Actor) wrapFastStop() { if !actor.isFastStop { return } startFunc := actor.startFunc actor.startFunc = func(ctx context.Context) error { errChan := make(chan error, 1) go func() { var err error defer func() { if v := recover(); v != nil { err = fmt.Errorf("recover: %s", v) } errChan <- err }() err = startFunc(ctx) }() select { case <-ctx.Done(): return ctx.Err() case err := <-errChan: return err } } } // start 同步启动 [Actor], [Actor].startFunc panic 或返回的 error 会被写入 [Actor].errorChan. func (actor *Actor) start() { actor.wrapFastStop() var err error defer func() { if v := recover(); v != nil { err = fmt.Errorf("recover: %s", v) } actor.group.errorsChan <- actorError{ actor: actor, err: err, } log.Tracef(context.Background(), "group-actor %s return", actor.name) close(actor.closeAfterStop) }() err = actor.startFunc(actor.startCtx) } // sendStopSignal 通知 [Actor] 退出. func (actor *Actor) sendStopSignal() { actor.startCancel() if actor.stopTimeout >= 0 { go func() { timer := time.NewTimer(actor.stopTimeout) select { case <-actor.closeAfterStop: timer.Stop() case <-timer.C: close(actor.closeAfterTimeout) } }() } } // wait 等待 [Actor] 停止或超时. func (actor *Actor) wait() { select { case <-actor.closeAfterTimeout: case <-actor.closeAfterStop: } } // WithName 指定 [Actor] 的 name; func (actor *Actor) WithName(name string) *Actor { actor.name = name return actor } // WithStartFunc 指定 [Actor] 的 startFunc; // 通过 WithStartFunc 注册的 startFunc 函数应该受 ctx 控制退出; func (actor *Actor) WithStartFunc(startFunc func(ctx context.Context) error) *Actor { actor.startFunc = startFunc return actor } // WithFastStop 设置 [Actor] 接收停止信号后立即终止. func (actor *Actor) WithFastStop(fastStop bool) *Actor { actor.isFastStop = fastStop return actor } // WithStopTimeout 指定停止 [Actor] 的最长等待时间, 设置 <0 没有最长等待时间. func (actor *Actor) WithStopTimeout(duration time.Duration) *Actor { actor.stopTimeout = duration return actor }