worker

Documentation

Index

• type ActivityTaskWorker
  • func (atw *ActivityTaskWorker) Complete(ctx context.Context, result *history.Event, task *backend.ActivityTask) error
  • func (atw *ActivityTaskWorker) Execute(ctx context.Context, task *backend.ActivityTask) (*history.Event, error)
  • func (atw *ActivityTaskWorker) Extend(ctx context.Context, task *backend.ActivityTask) error
  • func (atw *ActivityTaskWorker) Get(ctx context.Context, queues []workflow.Queue) (*backend.ActivityTask, error)
  • func (atw *ActivityTaskWorker) Start(ctx context.Context, queues []workflow.Queue) error
• type TaskWorker
• type Worker
  • func NewActivityWorker(b backend.Backend, registry *registry.Registry, clock clock.Clock, ...) *Worker[backend.ActivityTask, history.Event]
  • func NewWorker[Task, TaskResult any](b backend.Backend, tw TaskWorker[Task, TaskResult], options *WorkerOptions) *Worker[Task, TaskResult]
  • func NewWorkflowWorker(b backend.Backend, registry *registry.Registry, options WorkflowWorkerOptions) *Worker[backend.WorkflowTask, executor.ExecutionResult]
  • func (w *Worker[Task, TaskResult]) Start(ctx context.Context) error
  • func (w *Worker[Task, TaskResult]) WaitForCompletion() error
• type WorkerOptions
• type WorkflowTaskWorker
  • func (wtw *WorkflowTaskWorker) Complete(ctx context.Context, result *executor.ExecutionResult, t *backend.WorkflowTask) error
  • func (wtw *WorkflowTaskWorker) Execute(ctx context.Context, t *backend.WorkflowTask) (*executor.ExecutionResult, error)
  • func (wtw *WorkflowTaskWorker) Extend(ctx context.Context, t *backend.WorkflowTask) error
  • func (wtw *WorkflowTaskWorker) Get(ctx context.Context, queues []workflow.Queue) (*backend.WorkflowTask, error)
  • func (wtw *WorkflowTaskWorker) Start(ctx context.Context, queues []workflow.Queue) error
• type WorkflowWorkerOptions

Types

type ActivityTaskWorker

type ActivityTaskWorker struct {
	// contains filtered or unexported fields
}

func (*ActivityTaskWorker) Complete

func (atw *ActivityTaskWorker) Complete(ctx context.Context, result *history.Event, task *backend.ActivityTask) error

func (*ActivityTaskWorker) Execute

func (atw *ActivityTaskWorker) Execute(ctx context.Context, task *backend.ActivityTask) (*history.Event, error)

func (*ActivityTaskWorker) Extend

func (atw *ActivityTaskWorker) Extend(ctx context.Context, task *backend.ActivityTask) error

func (*ActivityTaskWorker) Get

func (atw *ActivityTaskWorker) Get(ctx context.Context, queues []workflow.Queue) (*backend.ActivityTask, error)

func (*ActivityTaskWorker) Start

func (atw *ActivityTaskWorker) Start(ctx context.Context, queues []workflow.Queue) error

type TaskWorker

type TaskWorker[Task, Result any] interface {
	Start(context.Context, []workflow.Queue) error
	Get(context.Context, []workflow.Queue) (*Task, error)
	Extend(context.Context, *Task) error
	Execute(context.Context, *Task) (*Result, error)
	Complete(context.Context, *Result, *Task) error
}

type Worker

type Worker[Task, TaskResult any] struct {
	// contains filtered or unexported fields
}

func NewActivityWorker

func NewActivityWorker(
	b backend.Backend,
	registry *registry.Registry,
	clock clock.Clock,
	options WorkerOptions,
) *Worker[backend.ActivityTask, history.Event]

func NewWorker

func NewWorker[Task, TaskResult any](
	b backend.Backend, tw TaskWorker[Task, TaskResult], options *WorkerOptions,
) *Worker[Task, TaskResult]

func NewWorkflowWorker

func NewWorkflowWorker(
	b backend.Backend,
	registry *registry.Registry,
	options WorkflowWorkerOptions,
) *Worker[backend.WorkflowTask, executor.ExecutionResult]

func (*Worker[Task, TaskResult]) Start

func (w *Worker[Task, TaskResult]) Start(ctx context.Context) error

func (*Worker[Task, TaskResult]) WaitForCompletion

func (w *Worker[Task, TaskResult]) WaitForCompletion() error

type WorkerOptions

type WorkerOptions struct {
	Pollers int

	MaxParallelTasks int

	HeartbeatInterval time.Duration

	PollingInterval time.Duration

	Queues []workflow.Queue
}

type WorkflowTaskWorker

type WorkflowTaskWorker struct {
	// contains filtered or unexported fields
}

func (*WorkflowTaskWorker) Complete

func (wtw *WorkflowTaskWorker) Complete(ctx context.Context, result *executor.ExecutionResult, t *backend.WorkflowTask) error

Complete implements TaskWorker.

func (*WorkflowTaskWorker) Execute

func (wtw *WorkflowTaskWorker) Execute(ctx context.Context, t *backend.WorkflowTask) (*executor.ExecutionResult, error)

func (*WorkflowTaskWorker) Extend

func (wtw *WorkflowTaskWorker) Extend(ctx context.Context, t *backend.WorkflowTask) error

func (*WorkflowTaskWorker) Get

func (wtw *WorkflowTaskWorker) Get(ctx context.Context, queues []workflow.Queue) (*backend.WorkflowTask, error)

func (*WorkflowTaskWorker) Start

func (wtw *WorkflowTaskWorker) Start(ctx context.Context, queues []workflow.Queue) error

type WorkflowWorkerOptions

type WorkflowWorkerOptions struct {
	WorkerOptions

	WorkflowExecutorCache     executor.Cache
	WorkflowExecutorCacheSize int
	WorkflowExecutorCacheTTL  time.Duration
}