crate/exegroup/actor.go

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
}