workflow

Documentation

Index

• func NewWorkflow(workflowFn reflect.Value) *workflow
• type ErrActivityAlreadyRegistered
  • func (e *ErrActivityAlreadyRegistered) Error() string
• type ErrInvalidActivity
  • func (e *ErrInvalidActivity) Error() string
• type ErrInvalidWorkflow
  • func (e *ErrInvalidWorkflow) Error() string
• type ErrWorkflowAlreadyRegistered
  • func (e *ErrWorkflowAlreadyRegistered) Error() string
• type ExecutionResult
• type ExecutorCache
• type RegisterConfig
• type RegisterOption
  • func WithName(name string) RegisterOption
• type Registry
  • func NewRegistry() *Registry
  • func (r *Registry) GetActivity(name string) (interface{}, error)
  • func (r *Registry) GetWorkflow(name string) (wf.Workflow, error)
  • func (r *Registry) RegisterActivity(activity wf.Activity, opts ...RegisterOption) error
  • func (r *Registry) RegisterWorkflow(workflow wf.Workflow, opts ...RegisterOption) error
• type WorkflowExecutor
  • func NewExecutor(logger *slog.Logger, tracer trace.Tracer, registry *Registry, ...) (WorkflowExecutor, error)
• type WorkflowHistoryProvider

Functions

func NewWorkflow

func NewWorkflow(workflowFn reflect.Value) *workflow

Types

type ErrActivityAlreadyRegistered

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

func (*ErrActivityAlreadyRegistered) Error

func (e *ErrActivityAlreadyRegistered) Error() string

type ErrInvalidActivity

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

func (*ErrInvalidActivity) Error

func (e *ErrInvalidActivity) Error() string

type ErrInvalidWorkflow

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

func (*ErrInvalidWorkflow) Error

func (e *ErrInvalidWorkflow) Error() string

type ErrWorkflowAlreadyRegistered

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

func (*ErrWorkflowAlreadyRegistered) Error

func (e *ErrWorkflowAlreadyRegistered) Error() string

type ExecutionResult

type ExecutionResult struct {
	State          core.WorkflowInstanceState
	Executed       []*history.Event
	ActivityEvents []*history.Event
	TimerEvents    []*history.Event
	WorkflowEvents []history.WorkflowEvent
}

type ExecutorCache

type ExecutorCache interface {
	Store(ctx context.Context, instance *core.WorkflowInstance, workflow WorkflowExecutor) error
	Get(ctx context.Context, instance *core.WorkflowInstance) (WorkflowExecutor, bool, error)
	StartEviction(ctx context.Context)
}

type RegisterConfig

type RegisterConfig struct {
	Name string
}

type RegisterOption

type RegisterOption interface {
	// contains filtered or unexported methods
}

func WithName

func WithName(name string) RegisterOption

type Registry

type Registry struct {
	sync.Mutex
	// contains filtered or unexported fields
}

func NewRegistry

func NewRegistry() *Registry

func (*Registry) GetActivity

func (r *Registry) GetActivity(name string) (interface{}, error)

func (*Registry) GetWorkflow

func (r *Registry) GetWorkflow(name string) (wf.Workflow, error)

func (*Registry) RegisterActivity

func (r *Registry) RegisterActivity(activity wf.Activity, opts ...RegisterOption) error

func (*Registry) RegisterWorkflow

func (r *Registry) RegisterWorkflow(workflow wf.Workflow, opts ...RegisterOption) error

type WorkflowExecutor

type WorkflowExecutor interface {
	ExecuteTask(ctx context.Context, t *backend.WorkflowTask) (*ExecutionResult, error)

	Close()
}

func NewExecutor

func NewExecutor(
	logger *slog.Logger,
	tracer trace.Tracer,
	registry *Registry,
	cv converter.Converter,
	propagators []wf.ContextPropagator,
	historyProvider WorkflowHistoryProvider,
	instance *core.WorkflowInstance,
	metadata *metadata.WorkflowMetadata,
	clock clock.Clock,
) (WorkflowExecutor, error)

type WorkflowHistoryProvider

type WorkflowHistoryProvider interface {
	GetWorkflowInstanceHistory(ctx context.Context, instance *core.WorkflowInstance, lastSequenceID *int64) ([]*history.Event, error)
}