workflow

Documentation

Index

• type Engine
  • func NewEngine(log *zap.Logger, ns NatsService) (*Engine, error)
  • func (c *Engine) CancelWorkflowInstance(ctx context.Context, id string, state model.CancellationState, ...) error
  • func (c *Engine) CompleteManualTask(ctx context.Context, trackingID string, newvars []byte) error
  • func (c *Engine) CompleteSendMessageTask(ctx context.Context, trackingID string, newvars []byte) error
  • func (c *Engine) CompleteServiceTask(ctx context.Context, trackingID string, newvars []byte) error
  • func (c *Engine) CompleteUserTask(ctx context.Context, trackingID string, newvars []byte) error
  • func (c *Engine) Launch(ctx context.Context, workflowName string, vars []byte) (string, error)
  • func (c *Engine) LoadWorkflow(ctx context.Context, model *model.Workflow) (string, error)
  • func (c *Engine) PublishError(ctx context.Context, state *model.WorkflowState, err error) (bool, error)
  • func (c *Engine) Shutdown()
  • func (c *Engine) Start(ctx context.Context) error
• type NatsService

Types

type Engine

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

Engine contains the workflow processing functions

func NewEngine

func NewEngine(log *zap.Logger, ns NatsService) (*Engine, error)

NewEngine returns an instance of the core workflow engine.

func (*Engine) CancelWorkflowInstance

func (c *Engine) CancelWorkflowInstance(ctx context.Context, id string, state model.CancellationState, wfError *model.Error) error

CancelWorkflowInstance will cancel a workflow instance with a reason

func (*Engine) CompleteManualTask

func (c *Engine) CompleteManualTask(ctx context.Context, trackingID string, newvars []byte) error

func (*Engine) CompleteSendMessageTask

func (c *Engine) CompleteSendMessageTask(ctx context.Context, trackingID string, newvars []byte) error

func (*Engine) CompleteServiceTask

func (c *Engine) CompleteServiceTask(ctx context.Context, trackingID string, newvars []byte) error

func (*Engine) CompleteUserTask

func (c *Engine) CompleteUserTask(ctx context.Context, trackingID string, newvars []byte) error

func (*Engine) Launch

func (c *Engine) Launch(ctx context.Context, workflowName string, vars []byte) (string, error)

Launch starts a new instance of a workflow and returns a workflow instance Id.

func (*Engine) LoadWorkflow

func (c *Engine) LoadWorkflow(ctx context.Context, model *model.Workflow) (string, error)

LoadWorkflow loads a model.Process describing a workflow into the engine ready for execution.

func (*Engine) PublishError

func (c *Engine) PublishError(ctx context.Context, state *model.WorkflowState, err error) (bool, error)

func (*Engine) Shutdown

func (c *Engine) Shutdown()

func (c *Engine) fatalEngineErr(msg string, err error, z ...zap.Field) error {
	return &errors.ErrWorkflowFatal{Err: c.engineErr(msg, err, z...)}
}

func (*Engine) Start

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

Start sets up the activity and job processors and starts the engine processing workflows.

type NatsService

type NatsService interface {
	AwaitMsg(ctx context.Context, state *model.WorkflowState) error
	SetTraversalProvider(provider services.TraversalFunc)
	ListWorkflows(ctx context.Context) (chan *model.ListWorkflowResult, chan error)
	StoreWorkflow(ctx context.Context, wf *model.Workflow) (string, error)
	GetWorkflow(ctx context.Context, workflowId string) (*model.Workflow, error)
	CreateWorkflowInstance(ctx context.Context, wfInstance *model.WorkflowInstance) (*model.WorkflowInstance, error)
	GetWorkflowInstance(ctx context.Context, workflowInstanceId string) (*model.WorkflowInstance, error)
	DestroyWorkflowInstance(ctx context.Context, workflowInstanceId string, state model.CancellationState, wfError *model.Error) error
	GetServiceTaskRoutingKey(taskName string) (string, error)
	GetMessageSenderRoutingKey(workflowName string, messageName string) (string, error)
	GetLatestVersion(ctx context.Context, workflowName string) (string, error)
	CreateJob(ctx context.Context, job *model.WorkflowState) (string, error)
	GetJob(ctx context.Context, id string) (*model.WorkflowState, error)
	GetElement(ctx context.Context, state *model.WorkflowState) (*model.Element, error)
	ListWorkflowInstance(ctx context.Context, workflowName string) (chan *model.ListWorkflowInstanceResult, chan error)
	GetWorkflowInstanceStatus(ctx context.Context, id string) (*model.WorkflowInstanceStatus, error)
	StartProcessing(ctx context.Context) error
	SetEventProcessor(processor services.EventProcessorFunc)
	SetMessageCompleteProcessor(processor services.MessageCompleteProcessorFunc)
	SetMessageProcessor(processor services.MessageProcessorFunc)
	SetCompleteJobProcessor(processor services.CompleteJobProcessorFunc)
	SetCompleteActivity(processor services.CompleteActivityFunc)
	SetAbort(processor services.AbortFunc)
	DeleteJob(ctx context.Context, trackingID string) error
	SetCompleteActivityProcessor(processor services.CompleteActivityProcessorFunc)
	SetLaunchFunc(processor services.LaunchFunc)
	PublishWorkflowState(ctx context.Context, stateName string, state *model.WorkflowState, ops ...services.PublishOpt) error
	PublishMessage(ctx context.Context, workflowInstanceID string, name string, key string, vars []byte) error
	Conn() common.NatsConn
	Shutdown()
	CloseUserTask(ctx context.Context, trackingID string) error
	OwnerId(name string) (string, error)
	OwnerName(id string) (string, error)
	GetOldState(id string) (*model.WorkflowState, error)
}