Documentation
¶
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func StartAsyncWorkers ¶
func StartAsyncWorkers( ctx context.Context, worker *AsyncWorker, n int, )
func StartAwaitPollWorkers ¶
func StartAwaitPollWorkers(ctx context.Context, worker *AwaitPollWorker, n int)
Types ¶
type AsyncEmitter ¶
type AsyncEmitter struct {
// contains filtered or unexported fields
}
func (*AsyncEmitter) EmitToolEvent ¶
func (e *AsyncEmitter) EmitToolEvent(event tool.ToolEvent)
type AsyncWorker ¶
type AsyncWorker struct {
// contains filtered or unexported fields
}
AsyncWorker 异步节点事件调度器
func NewAsyncWorker ¶
func NewAsyncWorker( queue engine.AsyncJobQueue, taskRepo repository.TaskRepository, nodeRuntimeRepo repository.NodeRuntimeRepository, tools *tool.Registry, eventBus *eventbus.EventBus, ) *AsyncWorker
type AwaitPollWorker ¶
type AwaitPollWorker struct {
// contains filtered or unexported fields
}
func NewAwaitPollWorker ¶
func NewAwaitPollWorker( awaitBindingRepo repository.AwaitBindingRepository, tools *tool.Registry, completer awaitCompleter, eventBus *eventbus.EventBus, dLocker lock.DistributedLock, ) *AwaitPollWorker
func (*AwaitPollWorker) RunOnce ¶
func (w *AwaitPollWorker) RunOnce(ctx context.Context) (int, error)
func (*AwaitPollWorker) Start ¶
func (w *AwaitPollWorker) Start(ctx context.Context)
type RecoveryScanner ¶
type RecoveryScanner struct {
// contains filtered or unexported fields
}
RecoveryScanner 数据库级恢复扫描器 Recovery Scanner（数据库级恢复扫描器）
负责：
- 扫描 DB 中 status = running 的任务
- 找出其中 node_state = running 的节点
- 把它们改为 retrying
- 把任务重新 push 进队列
- 保证不会重复恢复
func NewRecoveryScanner ¶
func NewRecoveryScanner( taskRepo repository.TaskRepository, nodeRepo repository.NodeRuntimeRepository, retryService engine.TaskRetryService, eventBus *eventbus.EventBus, interval time.Duration, timeout time.Duration, ) *RecoveryScanner
func (*RecoveryScanner) Start ¶
func (r *RecoveryScanner) Start(ctx context.Context)
type Worker ¶
type Worker struct {
// contains filtered or unexported fields
}
func NewWorker ¶
func NewWorker( eng *engine.Engine, taskRepo repository.TaskRepository, nodeRunTimeRepo repository.NodeRuntimeRepository, workflowVersionRepo repository.WorkflowVersionRepository, workflowRepo repository.WorkflowRepository, queue repository.TaskQueue, asyncJobQueue engine.AsyncJobQueue, eventBus *eventbus.EventBus, nodeRegistry *nodes.NodeRegistry, dLocker lock.DistributedLock, builder *workflow.Builder, taskRetryService engine.TaskRetryService, ) *Worker
func (*Worker) StartRecoveryScanner ¶
StartRecoveryScanner 启动数据库级恢复扫描器（后台 goroutine，ctx 取消时退出）。 由调用方显式启动，NewWorker 不再隐式拉起。
Source Files
Click to show internal directories.
Click to hide internal directories.