engine

Documentation

Overview

Package engine provides the core workflow engine implementation.

Index

• Variables
• type ActivityFunc
• type Config
  • func DefaultConfig() Config
  • func (c Config) Validate() error
• type Engine
  • func NewEngine(cfg Config) (*Engine, error)
  • func (e *Engine) CancelWorkflow(ctx context.Context, workflowID, runID string) error
  • func (e *Engine) Client() client.Client
  • func (e *Engine) DescribeWorkflow(ctx context.Context, workflowID, runID string) (*WorkflowExecutionInfo, error)
  • func (e *Engine) ExecuteWorkflow(ctx context.Context, workflowID string, workflow interface{}, ...) (client.WorkflowRun, error)
  • func (e *Engine) GetWorkflowHistory(ctx context.Context, workflowID, runID string) (client.HistoryEventIterator, error)
  • func (e *Engine) GetWorkflowResult(ctx context.Context, workflowID, runID string, result interface{}) error
  • func (e *Engine) IsRunning() bool
  • func (e *Engine) QueryWorkflow(ctx context.Context, workflowID, runID, queryType string, args ...interface{}) (interface{}, error)
  • func (e *Engine) RegisterActivity(act ActivityFunc)
  • func (e *Engine) RegisterWorkflow(wf WorkflowFunc)
  • func (e *Engine) SendSignal(ctx context.Context, workflowID, signalName string, signalData interface{}) error
  • func (e *Engine) Start(ctx context.Context) error
  • func (e *Engine) Stop() error
  • func (e *Engine) TerminateWorkflow(ctx context.Context, workflowID, runID, reason string) error
• type ErrActivityFailed
  • func (e ErrActivityFailed) Error() string
  • func (e ErrActivityFailed) Unwrap() error
• type ErrConfigInvalid
  • func (e ErrConfigInvalid) Error() string
• type ErrWorkflowFailed
  • func (e ErrWorkflowFailed) Error() string
  • func (e ErrWorkflowFailed) Unwrap() error
• type WorkflowExecutionInfo
• type WorkflowFunc

Variables

var (
	// ErrEngineNotStarted is returned when engine operations are called before Start.
	ErrEngineNotStarted = errors.New("workflow engine not started")
	// ErrEngineAlreadyStarted is returned when Start is called on a running engine.
	ErrEngineAlreadyStarted = errors.New("workflow engine already started")
	// ErrWorkflowNotFound is returned when a workflow execution is not found.
	ErrWorkflowNotFound = errors.New("workflow not found")
)

Types

type ActivityFunc

type ActivityFunc interface{}

ActivityFunc represents an activity function that can be registered.

type Config

type Config struct {
	// TemporalHostPort is the Temporal server address.
	TemporalHostPort string
	// Namespace is the Temporal namespace.
	Namespace string
	// TaskQueue is the task queue name for workflows and activities.
	TaskQueue string
	// MaxConcurrentWorkflows is the maximum number of concurrent workflow executions.
	MaxConcurrentWorkflows int
	// MaxConcurrentActivities is the maximum number of concurrent activity executions.
	MaxConcurrentActivities int
	// DefaultTimeout is the default workflow execution timeout.
	DefaultTimeout time.Duration
	// WorkerID is the unique identifier for this worker.
	WorkerID string
}

Config holds the workflow engine configuration.

func DefaultConfig

func DefaultConfig() Config

DefaultConfig returns a Config with sensible defaults.

func (Config) Validate

func (c Config) Validate() error

Validate validates the configuration.

type Engine

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

Engine orchestrates workflow execution using Temporal.

func NewEngine

func NewEngine(cfg Config) (*Engine, error)

NewEngine creates a new workflow engine with the given configuration.

func (*Engine) CancelWorkflow

func (e *Engine) CancelWorkflow(ctx context.Context, workflowID, runID string) error

CancelWorkflow cancels a running workflow execution.

func (*Engine) Client

func (e *Engine) Client() client.Client

Client returns the underlying Temporal client.

func (*Engine) DescribeWorkflow

func (e *Engine) DescribeWorkflow(ctx context.Context, workflowID, runID string) (*WorkflowExecutionInfo, error)

DescribeWorkflow returns information about a workflow execution.

func (*Engine) ExecuteWorkflow

func (e *Engine) ExecuteWorkflow(ctx context.Context, workflowID string, workflow interface{}, input interface{}) (client.WorkflowRun, error)

ExecuteWorkflow starts a new workflow execution.

func (*Engine) GetWorkflowHistory

func (e *Engine) GetWorkflowHistory(ctx context.Context, workflowID, runID string) (client.HistoryEventIterator, error)

GetWorkflowHistory retrieves the history of a workflow execution.

func (*Engine) GetWorkflowResult

func (e *Engine) GetWorkflowResult(ctx context.Context, workflowID, runID string, result interface{}) error

GetWorkflowResult retrieves the result of a workflow execution.

func (*Engine) IsRunning

func (e *Engine) IsRunning() bool

IsRunning returns true if the engine is currently running.

func (*Engine) QueryWorkflow

func (e *Engine) QueryWorkflow(ctx context.Context, workflowID, runID, queryType string, args ...interface{}) (interface{}, error)

QueryWorkflow queries a workflow for its current state.

func (*Engine) RegisterActivity

func (e *Engine) RegisterActivity(act ActivityFunc)

RegisterActivity registers an activity function with the engine.

func (*Engine) RegisterWorkflow

func (e *Engine) RegisterWorkflow(wf WorkflowFunc)

RegisterWorkflow registers a workflow function with the engine.

func (*Engine) SendSignal

func (e *Engine) SendSignal(ctx context.Context, workflowID, signalName string, signalData interface{}) error

SendSignal sends a signal to a running workflow.

func (*Engine) Start

func (e *Engine) Start(ctx context.Context) error

Start initializes the Temporal client and worker, then starts processing.

func (*Engine) Stop

func (e *Engine) Stop() error

Stop gracefully shuts down the engine.

func (*Engine) TerminateWorkflow

func (e *Engine) TerminateWorkflow(ctx context.Context, workflowID, runID, reason string) error

TerminateWorkflow forcefully terminates a workflow execution.

type ErrActivityFailed

type ErrActivityFailed struct {
	ActivityType string
	Cause        error
}

ErrActivityFailed is returned when an activity execution fails.

func (ErrActivityFailed) Error

func (e ErrActivityFailed) Error() string

func (ErrActivityFailed) Unwrap

func (e ErrActivityFailed) Unwrap() error

type ErrConfigInvalid

type ErrConfigInvalid struct {
	Field  string
	Reason string
}

ErrConfigInvalid is returned when configuration validation fails.

func (ErrConfigInvalid) Error

func (e ErrConfigInvalid) Error() string

type ErrWorkflowFailed

type ErrWorkflowFailed struct {
	WorkflowID string
	RunID      string
	Cause      error
}

ErrWorkflowFailed is returned when a workflow execution fails.

func (ErrWorkflowFailed) Error

func (e ErrWorkflowFailed) Error() string

func (ErrWorkflowFailed) Unwrap

func (e ErrWorkflowFailed) Unwrap() error

type WorkflowExecutionInfo

type WorkflowExecutionInfo struct {
	WorkflowID string
	RunID      string
	Type       string
	Status     string
}

WorkflowExecutionInfo contains information about a workflow execution.

type WorkflowFunc

type WorkflowFunc interface{}

WorkflowFunc represents a workflow function that can be registered.