worker

Documentation

Index

• Variables
• type ActivityWorkerOptions
• type Options
• type Worker
  • func New(backend backend.Backend, options *Options) *Worker
  • func NewActivityWorker(backend backend.Backend, options *ActivityWorkerOptions) *Worker
  • func NewWorkflowWorker(backend backend.Backend, options *WorkflowWorkerOptions) *Worker
  • func (w *Worker) RegisterActivity(a workflow.Activity, opts ...registry.RegisterOption) error
  • func (w *Worker) RegisterWorkflow(wf workflow.Workflow, opts ...registry.RegisterOption) error
  • func (w *Worker) Start(ctx context.Context) error
  • func (w *Worker) WaitForCompletion() error
• type WorkflowWorkerOptions

Variables

var DefaultOptions = Options{
	WorkflowWorkerOptions: WorkflowWorkerOptions{
		WorkflowPollers:           2,
		WorkflowPollingInterval:   200 * time.Millisecond,
		MaxParallelWorkflowTasks:  0,
		WorkflowHeartbeatInterval: 25 * time.Second,

		WorkflowExecutorCacheSize: 128,
		WorkflowExecutorCacheTTL:  time.Second * 10,
		WorkflowExecutorCache:     nil,
	},

	ActivityWorkerOptions: ActivityWorkerOptions{
		ActivityPollers:           2,
		ActivityPollingInterval:   200 * time.Millisecond,
		MaxParallelActivityTasks:  0,
		ActivityHeartbeatInterval: 25 * time.Second,
	},
}

Types

type ActivityWorkerOptions

type ActivityWorkerOptions struct {
	// ActivityPollers is the number of pollers to start. Defaults to 2.
	ActivityPollers int

	// MaxParallelActivityTasks determines the maximum number of concurrent activity tasks processed
	// by the worker. The default is 0 which is no limit.
	MaxParallelActivityTasks int

	// ActivityHeartbeatInterval is the interval between heartbeat attempts for activity tasks. Defaults
	// to 25 seconds
	ActivityHeartbeatInterval time.Duration

	// ActivityPollingInterval is the interval between polling for new activity tasks.
	// Note that if you use a backend that can wait for tasks to be available (e.g. redis) this field has no effect.
	// Defaults to 200ms.
	ActivityPollingInterval time.Duration

	// ActivityQueues are the queues the worker listens to
	ActivityQueues []workflow.Queue
}

type Options

type Options struct {
	WorkflowWorkerOptions
	ActivityWorkerOptions
}

type Worker

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

func New

func New(backend backend.Backend, options *Options) *Worker

New creates a worker that processes workflows and activities.

func NewActivityWorker

func NewActivityWorker(backend backend.Backend, options *ActivityWorkerOptions) *Worker

NewActivityWorker creates a worker that only processes activities.

func NewWorkflowWorker

func NewWorkflowWorker(backend backend.Backend, options *WorkflowWorkerOptions) *Worker

NewWorkflowWorker creates a worker that only processes workflows.

func (*Worker) RegisterActivity

func (w *Worker) RegisterActivity(a workflow.Activity, opts ...registry.RegisterOption) error

RegisterActivity registers an activity with the worker's registry.

func (*Worker) RegisterWorkflow

func (w *Worker) RegisterWorkflow(wf workflow.Workflow, opts ...registry.RegisterOption) error

RegisterWorkflow registers a workflow with the worker's registry.

func (*Worker) Start

func (w *Worker) Start(ctx context.Context) error

Start starts the worker.

To stop the worker, cancel the context passed to Start. To wait for completion of the active tasks, call `WaitForCompletion`.

func (*Worker) WaitForCompletion

func (w *Worker) WaitForCompletion() error

WaitForCompletion waits for all active tasks to complete.

type WorkflowWorkerOptions

type WorkflowWorkerOptions struct {
	// WorkflowsPollers is the number of pollers to start. Defaults to 2.
	WorkflowPollers int

	// MaxParallelWorkflowTasks determines the maximum number of concurrent workflow tasks processed
	// by the worker. The default is 0 which is no limit.
	MaxParallelWorkflowTasks int

	// WorkflowHeartbeatInterval is the interval between heartbeat attempts on workflow tasks. Defaults
	// to 25 seconds
	WorkflowHeartbeatInterval time.Duration

	// WorkflowPollingInterval is the interval between polling for new workflow tasks.
	// Note that if you use a backend that can wait for tasks to be available (e.g. redis) this field has no effect.
	// Defaults to 200ms.
	WorkflowPollingInterval time.Duration

	// WorkflowExecutorCache is the max size of the workflow executor cache. Defaults to 128
	WorkflowExecutorCacheSize int

	// WorkflowExecutorCache is the max TTL of the workflow executor cache. Defaults to 10 seconds
	WorkflowExecutorCacheTTL time.Duration

	// WorkflowExecutorCache is the cache to use for workflow executors. If nil, a default cache implementation
	// will be used.
	WorkflowExecutorCache executor.Cache

	// WorkflowQueues are the queue the worker listens to
	WorkflowQueues []workflow.Queue
}