instance

package
v1.6.29 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 8, 2026 License: BSD-3-Clause Imports: 36 Imported by: 11

Documentation

Index

Constants

View Source 
const (
	OpStart   = iota // 0
	OpResume         // 1
	OpRestart        // 2
)
View Source 
const (
	CodeTxRolledBack   = "SUBFLOW-TX-001" // rollback - RETRIABLE *activity.Error (D14)
	CodeTxCommitFailed = "SUBFLOW-TX-003" // commit   - PLAIN error, never retried (D14)
	CodeTxSweep        = "SUBFLOW-TX-006"
	CodeTxNotResumable = "SUBFLOW-TX-007"
)

Error codes. CodeTxRolledBack is the ONLY finalisation error that is an *activity.Error.

View Source 
const EngineError string = "ENGINE-ERROR"
View Source 
const EventIdAttr = "event.id"
View Source 
const NumConversionError string = "NUM-CONVERSION-001"
View Source 
const SchemaValidationError string = "SCHEMA-VALIDATION-001"
View Source 
const TxInFlightAttr = "_txInFlight"

TxInFlightAttr marks an instance whose recorded state was captured while a transactional subflow was in flight. Instance.Delete is an empty function, so the marker is CLEARED by setting it to false, never by deleting it.

Variables

View Source 
var ErrTxDowngraded = errors.New("SUBFLOW-TX-001: commit downgraded to rollback: a task failed while the transaction was being committed")

ErrTxDowngraded is returned by Commit when confirm() vetoed the commit under the operation lock. The transaction was ROLLED BACK.

Functions

func ApplyExecOptions 

func ApplyExecOptions(instance *IndependentInstance, execOptions *ExecOptions)

ApplyExecOptions applies any execution options to the flow instance

func EnableChangeTracking added in v0.9.4 

func EnableChangeTracking(enable bool, mode state.RecordingMode)

func GetFlowIOMetadata 

func GetFlowIOMetadata(flowURI string) (*metadata.IOMetadata, error)

func IsConcurrentTaskExcutionEnabled added in v1.6.15 

func IsConcurrentTaskExcutionEnabled() bool

func IsTxDowngraded added in v1.6.29 

func IsTxDowngraded(err error) bool

IsTxDowngraded reports whether a Commit was downgraded to a rollback.

func NewErrorObj 

func NewErrorObj(taskId string, msg string) map[string]interface{}

func NewStateInstanceRecorder added in v1.6.0 

func NewStateInstanceRecorder(recorder state.Recorder, mod state.RecordingMode, rerunstate bool) *stateInstanceRecorder

func RejectIfTxInFlight added in v1.6.29 

func RejectIfTxInFlight(inst *IndependentInstance) error

RejectIfTxInFlight refuses to resume a flow whose recorded state was captured inside a transactional subflow. The transaction did not survive the restart, so resuming would run the remainder of the subflow with no transaction at all and then report success (D10).

func RollbackOpenTransactions added in v1.6.29 

func RollbackOpenTransactions(inst *IndependentInstance)

RollbackOpenTransactions rolls back any transaction still open when the flow stops running.

It is registered as the FIRST defer in each driver's goroutine so it runs LAST, surviving a panic in the tail and the detached branch, which has no defer of its own. Reaching it means a terminal transition was missed - the step loop hit its max-step cap, or the driver returned early - so it logs at ERROR rather than staying silent.

O(1) when the feature is unused: txScopeActive is zero and nothing is scanned.

func SetTxContextPropagator added in v1.6.29 

func SetTxContextPropagator(fn func(src, dst context.Context) context.Context)

SetTxContextPropagator registers the real propagator. Called from the subflow activity's init(); passing nil restores the identity default.

func StartDetachedSubFlow added in v1.6.8 

func StartDetachedSubFlow(ctx activity.Context, flowURI string, inputs map[string]interface{}) error

func StartSubFlow 

func StartSubFlow(ctx activity.Context, flowURI string, inputs map[string]interface{}) error

func StartSubFlowWithContext added in v1.6.21 

func StartSubFlowWithContext(duration int64, ctx activity.Context, flowURI string, inputs map[string]interface{}) error

func StartTransactionalSubFlow added in v1.6.29 

func StartTransactionalSubFlow(ctx activity.Context, flowURI string, inputs map[string]interface{},
	timeoutMs int64, connID string, decorate TxContextDecorator, fin TxFinalizer) error

StartTransactionalSubFlow starts an embedded subflow that runs inside a database transaction (FLOGO-19484). It is ADDITIVE: StartSubFlow, StartSubFlowWithContext and StartDetachedSubFlow are untouched, so an older build of the separately-versioned flow/activity/subflow module still compiles against a newer flow.

The ENGINE owns context construction. The activity supplies only `decorate`, which layers the transaction handle onto the context, and `fin`, which commits or rolls back; flow/instance never imports database/sql.

Types

type ActivityEvalError 

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

func NewActivityEvalError 

func NewActivityEvalError(taskName string, errorType string, errorText string) *ActivityEvalError

func (*ActivityEvalError) Error 

func (e *ActivityEvalError) Error() string

func (*ActivityEvalError) TaskName 

func (e *ActivityEvalError) TaskName() string

func (*ActivityEvalError) Type 

func (e *ActivityEvalError) Type() string

type ChangeTracker added in v0.9.4 

type ChangeTracker interface {
	// FlowCreated is called to track a when a flow is created
	FlowCreated(flow *IndependentInstance)
	// FlowDone is called to track a when a flow complete
	FlowDone(flow *IndependentInstance)
	// SetStatus is called to track a status change on an instance
	SetStatus(subflowId int, status model.FlowStatus)
	// AttrChange is called to track when Attribute changes
	AttrChange(subflowId int, name string, value interface{})
	// SubflowCreated is called to track a when a subflow is created
	SubflowCreated(subflow *Instance)
	// SubflowDone is called to track a when a subflow complete
	SubflowDone(subflow *Instance)
	// WorkItemAdded records when an item is added to the WorkQueue
	WorkItemAdded(wi *WorkItem)
	// WorkItemRemoved records when an item is removed from the WorkQueue
	WorkItemRemoved(wi *WorkItem)
	// TaskAdded records when a Task is added
	TaskAdded(taskInst *TaskInst)
	// TaskUpdated records when a Task is updated
	TaskUpdated(taskInst *TaskInst)
	// TaskRemoved records when a Task is removed
	TaskRemoved(subflowId int, taskId string)
	// LinkAdded records a Link is added
	LinkAdded(linkInst *LinkInst)
	// LinkUpdated records a Link is updated
	LinkUpdated(linkInst *LinkInst)
	// LinkRemoved records when a Link is removed
	LinkRemoved(subflowId int, linkId int)
	// ExtractStep extracts the step object and resets the tracker
	ExtractStep(reset bool) *state.Step
}

func NewInstanceChangeTracker 

func NewInstanceChangeTracker(flowId string, startStepId int) ChangeTracker

type ExecOptions 

type ExecOptions struct {
	Patch       *flowSupport.Patch
	Interceptor *support.Interceptor
	InstanceId  string
}

ExecOptions are optional Patch & Interceptor to be used during instance execution

type IDGenerator 

type IDGenerator interface {
	//NewFlowInstanceID generate a new instance ID
	NewFlowInstanceID() string
}

IDGenerator generates IDs for flow instances

type IDResponse 

type IDResponse struct {
	ID string `json:"id"`
}

IDResponse is a response object consists of an ID

type IndependentInstance 

type IndependentInstance struct {
	*Instance
	// contains filtered or unexported fields
}

func NewIndependentInstance 

func NewIndependentInstance(instanceID string, flowURI string, flow *definition.Definition, instRecorder *stateInstanceRecorder, logger log.Logger, ctx context.Context) (*IndependentInstance, error)

New creates a new Flow Instance from the specified Flow

func (*IndependentInstance) ApplyInterceptor 

func (inst *IndependentInstance) ApplyInterceptor(interceptor *coresupport.Interceptor)

func (*IndependentInstance) ApplyPatch 

func (inst *IndependentInstance) ApplyPatch(patch *flowsupport.Patch)

func (*IndependentInstance) CurrentStep added in v0.9.4 

func (inst *IndependentInstance) CurrentStep(reset bool) *state.Step

func (*IndependentInstance) DoStep 

func (inst *IndependentInstance) DoStep() bool

func (*IndependentInstance) ExecutionTime added in v1.4.0 

func (inst *IndependentInstance) ExecutionTime() time.Duration

func (*IndependentInstance) GetChanges 

func (inst *IndependentInstance) GetChanges() ChangeTracker

GetChanges returns the Change Tracker object

func (*IndependentInstance) GetFlowState added in v1.6.0 

func (inst *IndependentInstance) GetFlowState(inputs map[string]interface{}) *state.FlowState

func (*IndependentInstance) GetInterceptor added in v1.6.3 

func (inst *IndependentInstance) GetInterceptor() *coresupport.Interceptor

func (*IndependentInstance) GetTimeoutContext added in v1.6.21 

func (inst *IndependentInstance) GetTimeoutContext() context.Context

func (*IndependentInstance) HandleCancelError added in v1.6.21 

func (inst *IndependentInstance) HandleCancelError(containerInst *Instance, err error)

func (*IndependentInstance) HandleGlobalError 

func (inst *IndependentInstance) HandleGlobalError(containerInst *Instance, err error)

HandleGlobalError handles instance errors.

FLOGO-19484: callers that hold the instance state lock MUST use handleGlobalError(.., true) instead. This exported form assumes the lock is NOT held.

func (*IndependentInstance) HasInterceptor added in v1.6.12 

func (inst *IndependentInstance) HasInterceptor() bool

func (*IndependentInstance) MarshalJSON 

func (inst *IndependentInstance) MarshalJSON() ([]byte, error)

MarshalJSON overrides the default MarshalJSON for FlowInstance

func (*IndependentInstance) RecordState added in v1.6.0 

func (inst *IndependentInstance) RecordState(strtTime time.Time) error

func (*IndependentInstance) ResetChanges 

func (inst *IndependentInstance) ResetChanges()

ResetChanges resets an changes that were being tracked

func (*IndependentInstance) Restart 

func (inst *IndependentInstance) Restart(logger log.Logger, id string, initStepId int) error

// Restart indicates that this FlowInstance was restarted

func (*IndependentInstance) RunConcurrent added in v1.6.27 

func (inst *IndependentInstance) RunConcurrent(stepCount, maxStepCount int, stateRecorder state.Recorder) int

RunConcurrent drains the work-item queue using a bounded worker pool so that ready tasks (e.g. parallel transition branches) execute concurrently. It is used only when concurrent task execution is enabled; the sequential DoStep loop is left completely untouched. The returned value is the updated step count for the caller's max-step bookkeeping.

Failure policy is drain-then-fail: the first unhandled branch error is latched and the group context is cancelled so siblings abort; the pool then waits for all in-flight work to return before the (single) global error handler runs.

func (*IndependentInstance) SetInstanceRecorder added in v1.6.0 

func (inst *IndependentInstance) SetInstanceRecorder(stateRecorder *stateInstanceRecorder)

func (*IndependentInstance) SetTracingContext added in v0.9.4 

func (inst *IndependentInstance) SetTracingContext(tracingCtx trace.TracingContext)

func (*IndependentInstance) Snapshot added in v0.9.4 

func (inst *IndependentInstance) Snapshot() *state.Snapshot

func (*IndependentInstance) Start 

func (inst *IndependentInstance) Start(startAttrs map[string]interface{}) bool

func (*IndependentInstance) StepID 

func (inst *IndependentInstance) StepID() int

StepID returns the current step ID of the Flow Instance

func (*IndependentInstance) UnmarshalJSON 

func (inst *IndependentInstance) UnmarshalJSON(d []byte) error

UnmarshalJSON overrides the default UnmarshalJSON for FlowInstance

func (*IndependentInstance) UpdateStartTime added in v1.4.0 

func (inst *IndependentInstance) UpdateStartTime()

type Instance 

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

func (*Instance) Delete added in v1.6.20 

func (inst *Instance) Delete(name string)

func (*Instance) FindOrCreateLinkData 

func (inst *Instance) FindOrCreateLinkData(link *definition.Link) (linkInst *LinkInst, created bool)

FindOrCreateLinkData finds an existing LinkInst or creates ones if not found for the specified link the task environment

func (*Instance) FindOrCreateTaskInst added in v0.10.0 

func (inst *Instance) FindOrCreateTaskInst(task *definition.Task) (taskInst *TaskInst, created bool)

FindOrCreateTaskInst finds an existing TaskInst or creates ones if not found for the specified task the task environment

func (*Instance) FlowDefinition 

func (inst *Instance) FlowDefinition() *definition.Definition

FlowDefinition returns the Flow definition associated with this context

func (*Instance) FlowURI 

func (inst *Instance) FlowURI() string

func (*Instance) GetError 

func (inst *Instance) GetError() error

func (*Instance) GetMasterScope added in v1.3.0 

func (inst *Instance) GetMasterScope() data.Scope

func (*Instance) GetReturnData 

func (inst *Instance) GetReturnData() (map[string]interface{}, error)

func (*Instance) GetValue 

func (inst *Instance) GetValue(name string) (value interface{}, exists bool)

func (*Instance) ID 

func (inst *Instance) ID() string

func (*Instance) IOMetadata 

func (inst *Instance) IOMetadata() *metadata.IOMetadata

IOMetadata get the input/output metadata of the activity host

func (*Instance) Logger 

func (inst *Instance) Logger() log.Logger

func (*Instance) MarshalJSON 

func (inst *Instance) MarshalJSON() ([]byte, error)

MarshalJSON overrides the default MarshalJSON for FlowInstance

func (*Instance) Name 

func (inst *Instance) Name() string

func (*Instance) Reply 

func (inst *Instance) Reply(replyData map[string]interface{}, err error)

func (*Instance) Return 

func (inst *Instance) Return(returnData map[string]interface{}, err error)

func (*Instance) Scope 

func (inst *Instance) Scope() data.Scope

func (*Instance) SetResultHandler 

func (inst *Instance) SetResultHandler(handler action.ResultHandler)

InitActionContext initialize the action context, should be initialized before execution

func (*Instance) SetStatus 

func (inst *Instance) SetStatus(status model.FlowStatus)

func (*Instance) SetTriggerTags added in v1.6.27 

func (inst *Instance) SetTriggerTags(tags map[string]interface{})

func (*Instance) SetValue 

func (inst *Instance) SetValue(name string, value interface{}) error

func (*Instance) SpanConfig added in v0.9.4 

func (inst *Instance) SpanConfig() trace.Config

func (*Instance) Status 

func (inst *Instance) Status() model.FlowStatus

Status returns the current status of the Flow Instance

func (*Instance) TaskInstances 

func (inst *Instance) TaskInstances() []model.TaskInstance

TaskInstances get the task instances

func (*Instance) TracingContext added in v0.9.4 

func (inst *Instance) TracingContext() trace.TracingContext

func (*Instance) UnmarshalJSON 

func (inst *Instance) UnmarshalJSON(d []byte) error

UnmarshalJSON overrides the default UnmarshalJSON for FlowInstance

func (*Instance) UpdateAttrs 

func (inst *Instance) UpdateAttrs(attrs map[string]interface{})

UpdateAttrs updates the attributes of the Flow Instance

type LegacyCtx 

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

DEPRECATED

func (*LegacyCtx) ActivityHost 

func (l *LegacyCtx) ActivityHost() activity.Host

func (*LegacyCtx) GetInput 

func (l *LegacyCtx) GetInput(name string) interface{}

func (*LegacyCtx) GetInputObject 

func (l *LegacyCtx) GetInputObject(input data.StructValue) error

func (*LegacyCtx) GetInputSchema 

func (l *LegacyCtx) GetInputSchema(name string) schema.Schema

func (*LegacyCtx) GetOutput 

func (l *LegacyCtx) GetOutput(name string) interface{}

func (*LegacyCtx) GetOutputSchema 

func (l *LegacyCtx) GetOutputSchema(name string) schema.Schema

func (*LegacyCtx) GetSetting added in v0.9.2 

func (l *LegacyCtx) GetSetting(name string) (interface{}, bool)

func (*LegacyCtx) GetSharedTempData 

func (l *LegacyCtx) GetSharedTempData() map[string]interface{}

func (*LegacyCtx) GetTracingContext added in v0.9.4 

func (l *LegacyCtx) GetTracingContext() trace.TracingContext

func (*LegacyCtx) GoContext added in v1.6.21 

func (l *LegacyCtx) GoContext() context.Context

func (*LegacyCtx) Logger 

func (l *LegacyCtx) Logger() log.Logger

func (*LegacyCtx) Name 

func (l *LegacyCtx) Name() string

func (*LegacyCtx) SetOutput 

func (l *LegacyCtx) SetOutput(name string, value interface{}) error

func (*LegacyCtx) SetOutputObject 

func (l *LegacyCtx) SetOutputObject(output data.StructValue) error

type LinkInst 

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

LinkInst represents data associated with an instance of a Link

func NewLinkInst 

func NewLinkInst(inst *Instance, link *definition.Link) *LinkInst

NewLinkInst creates a LinkInst for the specified link in the specified task environment 

func (li *LinkInst) Link() *definition.Link

Link returns the Link associated with ld context

func (*LinkInst) MarshalJSON 

func (li *LinkInst) MarshalJSON() ([]byte, error)

MarshalJSON overrides the default MarshalJSON for LinkInst

func (*LinkInst) SetStatus 

func (li *LinkInst) SetStatus(status model.LinkStatus)

SetStatus sets the current state indicator for the LinkInst

func (*LinkInst) Status 

func (li *LinkInst) Status() model.LinkStatus

Status returns the current state indicator for the LinkInst

func (*LinkInst) UnmarshalJSON 

func (li *LinkInst) UnmarshalJSON(d []byte) error

UnmarshalJSON overrides the default UnmarshalJSON for LinkInst

type NoopChangeTracker added in v0.9.4 

type NoopChangeTracker struct {
}

func (*NoopChangeTracker) AttrChange added in v0.9.4 

func (nct *NoopChangeTracker) AttrChange(subflowId int, name string, value interface{})

func (*NoopChangeTracker) ExtractStep added in v0.9.4 

func (nct *NoopChangeTracker) ExtractStep(reset bool) *state.Step

func (*NoopChangeTracker) FlowCreated added in v0.10.0 

func (nct *NoopChangeTracker) FlowCreated(flow *IndependentInstance)

func (*NoopChangeTracker) FlowDone added in v1.1.0 

func (nct *NoopChangeTracker) FlowDone(flow *IndependentInstance)

func (*NoopChangeTracker) LinkAdded added in v0.9.4 

func (nct *NoopChangeTracker) LinkAdded(linkInst *LinkInst)

func (*NoopChangeTracker) LinkRemoved added in v0.9.4 

func (nct *NoopChangeTracker) LinkRemoved(subflowId int, linkId int)

func (*NoopChangeTracker) LinkUpdated added in v0.9.4 

func (nct *NoopChangeTracker) LinkUpdated(linkInst *LinkInst)

func (*NoopChangeTracker) SetStatus added in v0.9.4 

func (nct *NoopChangeTracker) SetStatus(subflowId int, status model.FlowStatus)

func (*NoopChangeTracker) SubflowCreated added in v0.9.4 

func (nct *NoopChangeTracker) SubflowCreated(subflow *Instance)

func (*NoopChangeTracker) SubflowDone added in v1.1.0 

func (nct *NoopChangeTracker) SubflowDone(subflow *Instance)

func (*NoopChangeTracker) TaskAdded added in v0.9.4 

func (nct *NoopChangeTracker) TaskAdded(taskInst *TaskInst)

func (*NoopChangeTracker) TaskRemoved added in v0.9.4 

func (nct *NoopChangeTracker) TaskRemoved(subflowId int, taskId string)

func (*NoopChangeTracker) TaskUpdated added in v0.9.4 

func (nct *NoopChangeTracker) TaskUpdated(taskInst *TaskInst)

func (*NoopChangeTracker) WorkItemAdded added in v0.9.4 

func (nct *NoopChangeTracker) WorkItemAdded(wi *WorkItem)

func (*NoopChangeTracker) WorkItemRemoved added in v0.9.4 

func (nct *NoopChangeTracker) WorkItemRemoved(wi *WorkItem)

type RunOptions 

type RunOptions struct {
	Op                  int
	ReturnID            bool
	FlowURI             string
	PreservedInstanceId string
	InitStepId          int
	InitialState        *IndependentInstance
	ExecOptions         *ExecOptions
	Rerun               bool
	OriginalInstanceId  string
	DetachExecution     bool
}

RunOptions the options when running a FlowAction

type SimpleChangeTracker added in v0.9.4 

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

func (*SimpleChangeTracker) AttrChange added in v0.9.4 

func (sct *SimpleChangeTracker) AttrChange(subflowId int, name string, value interface{})

func (*SimpleChangeTracker) ExtractStep added in v0.9.4 

func (sct *SimpleChangeTracker) ExtractStep(reset bool) *state.Step

func (*SimpleChangeTracker) FlowCreated added in v0.10.0 

func (sct *SimpleChangeTracker) FlowCreated(flow *IndependentInstance)

func (*SimpleChangeTracker) FlowDone added in v1.1.0 

func (sct *SimpleChangeTracker) FlowDone(flow *IndependentInstance)

func (*SimpleChangeTracker) LinkAdded added in v0.9.4 

func (sct *SimpleChangeTracker) LinkAdded(linkInst *LinkInst)

func (*SimpleChangeTracker) LinkRemoved added in v0.9.4 

func (sct *SimpleChangeTracker) LinkRemoved(subflowId int, linkId int)

func (*SimpleChangeTracker) LinkUpdated added in v0.9.4 

func (sct *SimpleChangeTracker) LinkUpdated(linkInst *LinkInst)

func (*SimpleChangeTracker) SetStatus added in v0.9.4 

func (sct *SimpleChangeTracker) SetStatus(subflowId int, status model.FlowStatus)

func (*SimpleChangeTracker) SubflowCreated added in v0.9.4 

func (sct *SimpleChangeTracker) SubflowCreated(subflow *Instance)

func (*SimpleChangeTracker) SubflowDone added in v1.1.0 

func (sct *SimpleChangeTracker) SubflowDone(subflow *Instance)

func (*SimpleChangeTracker) TaskAdded added in v0.9.4 

func (sct *SimpleChangeTracker) TaskAdded(taskInst *TaskInst)

func (*SimpleChangeTracker) TaskRemoved added in v0.9.4 

func (sct *SimpleChangeTracker) TaskRemoved(subflowId int, taskId string)

func (*SimpleChangeTracker) TaskUpdated added in v0.9.4 

func (sct *SimpleChangeTracker) TaskUpdated(taskInst *TaskInst)

func (*SimpleChangeTracker) WorkItemAdded added in v0.9.4 

func (sct *SimpleChangeTracker) WorkItemAdded(wi *WorkItem)

func (*SimpleChangeTracker) WorkItemRemoved added in v0.9.4 

func (sct *SimpleChangeTracker) WorkItemRemoved(wi *WorkItem)

type SimpleChangeTrackerFactory added in v0.9.4 

type SimpleChangeTrackerFactory struct {
}

func (*SimpleChangeTrackerFactory) NewChangeTracker added in v0.9.4 

func (sf *SimpleChangeTrackerFactory) NewChangeTracker(flowId string, mode state.RecordingMode, startStepId int) ChangeTracker

type TaskInst 

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

func NewTaskInst 

func NewTaskInst(flowInst *Instance, task *definition.Task) *TaskInst

func (*TaskInst) ActivityHost 

func (ti *TaskInst) ActivityHost() activity.Host

func (*TaskInst) EvalActivity 

func (ti *TaskInst) EvalActivity() (done bool, evalErr error)

EvalActivity implements activity.ActivityContext.EvalActivity method 

func (ti *TaskInst) EvalLink(link *definition.Link) (result bool, err error)

EvalLink implements activity.ActivityContext.EvalLink method

func (*TaskInst) FlowLogger 

func (ti *TaskInst) FlowLogger() log.Logger

func (*TaskInst) GetFromLinkInstances 

func (ti *TaskInst) GetFromLinkInstances() []model.LinkInstance

GetFromLinkInstances implements model.TaskContext.GetFromLinkInstances

func (*TaskInst) GetGoContext added in v1.6.21 

func (ti *TaskInst) GetGoContext() context.Context

func (*TaskInst) GetInput 

func (ti *TaskInst) GetInput(name string) interface{}

GetInput implements activity.Context.GetInput

func (*TaskInst) GetInputObject 

func (ti *TaskInst) GetInputObject(input data.StructValue) error

GetInputObject implements activity.Context.GetInputObject

func (*TaskInst) GetInputSchema 

func (ti *TaskInst) GetInputSchema(name string) schema.Schema

func (*TaskInst) GetOutputSchema 

func (ti *TaskInst) GetOutputSchema(name string) schema.Schema

func (*TaskInst) GetSetting 

func (ti *TaskInst) GetSetting(name string) (value interface{}, exists bool)

func (*TaskInst) GetSharedTempData 

func (ti *TaskInst) GetSharedTempData() map[string]interface{}

func (*TaskInst) GetToLinkInstances 

func (ti *TaskInst) GetToLinkInstances() []model.LinkInstance

GetToLinkInstances implements model.TaskContext.GetToLinkInstances,

func (*TaskInst) GetTracingContext added in v0.9.4 

func (ti *TaskInst) GetTracingContext() trace.TracingContext

GetInputObject implements activity.Context.GetInputObject

func (*TaskInst) GetWorkingData 

func (ti *TaskInst) GetWorkingData(key string) (interface{}, bool)

func (*TaskInst) GetWorkingDataScope added in v0.9.4 

func (ti *TaskInst) GetWorkingDataScope() *WorkingDataScope

func (*TaskInst) GoContext added in v1.6.21 

func (ti *TaskInst) GoContext() context.Context

func (*TaskInst) HasActivity 

func (ti *TaskInst) HasActivity() bool

HasActivity implements activity.ActivityContext.HasActivity method

func (*TaskInst) InstanceId added in v0.9.4 

func (ti *TaskInst) InstanceId() string

Returns task instance id

func (*TaskInst) Logger 

func (ti *TaskInst) Logger() log.Logger

func (*TaskInst) MarshalJSON 

func (ti *TaskInst) MarshalJSON() ([]byte, error)

MarshalJSON overrides the default MarshalJSON for TaskInst

func (*TaskInst) Name 

func (ti *TaskInst) Name() string

Name implements activity.Context.Name method

func (*TaskInst) PostEvalActivity 

func (ti *TaskInst) PostEvalActivity() (done bool, evalErr error)

EvalActivity implements activity.ActivityContext.EvalActivity method

func (*TaskInst) SetOutput 

func (ti *TaskInst) SetOutput(name string, value interface{}) error

SetOutput implements activity.Context.SetOutput

func (*TaskInst) SetOutputObject 

func (ti *TaskInst) SetOutputObject(output data.StructValue) error

SetOutputObject implements activity.Context.SetOutputObject

func (*TaskInst) SetOutputs added in v1.6.0 

func (ti *TaskInst) SetOutputs(outputs map[string]interface{}) error

func (*TaskInst) SetStatus 

func (ti *TaskInst) SetStatus(status model.TaskStatus)

SetStatus implements flow.TaskContext.SetStatus

func (*TaskInst) SetWorkingData 

func (ti *TaskInst) SetWorkingData(key string, value interface{})

func (*TaskInst) SpanConfig added in v0.9.4 

func (ti *TaskInst) SpanConfig() trace.Config

func (*TaskInst) Status 

func (ti *TaskInst) Status() model.TaskStatus

Status implements flow.TaskContext.GetState

func (*TaskInst) Task 

func (ti *TaskInst) Task() *definition.Task

Task implements model.TaskContext.Task, by returning the Task associated with this TaskInst object

func (*TaskInst) UnmarshalJSON 

func (ti *TaskInst) UnmarshalJSON(d []byte) error

UnmarshalJSON overrides the default UnmarshalJSON for TaskInst

func (*TaskInst) UpdateTaskToTracker added in v1.3.0 

func (ti *TaskInst) UpdateTaskToTracker()

UpdateTaskToTracker update task to tracker

type TxCommitError added in v1.6.29 

type TxCommitError struct {
	ConnID string
	Err    error
}

TxCommitError is returned when COMMIT itself failed. It is deliberately a PLAIN error and deliberately NOT an *activity.Error.

D14: a failed COMMIT is IN DOUBT - the server may have applied the transaction and lost the acknowledgement. Replaying the subflow could double-apply every write in it, plus any non-transactional side effect it performed. taskbehavior.go's PostEval path retries any *activity.Error WITHOUT consulting Retriable, so the only reliable way to stay un-retried is to not be one.

Do not "improve" this into an activity.NewActivityError; TestCommitFailureErrorIsNotAnActivityError exists to stop that.

func (*TxCommitError) Error added in v1.6.29 

func (e *TxCommitError) Error() string

func (*TxCommitError) Unwrap added in v1.6.29 

func (e *TxCommitError) Unwrap() error

type TxContextDecorator added in v1.6.29 

type TxContextDecorator func(context.Context) context.Context

TxContextDecorator layers the ambient transaction onto the context handed to the embedded subflow instance. Supplied by the subflow activity, which is the only component that knows about database/sql.

type TxFinalizer added in v1.6.29 

type TxFinalizer interface {
	Commit(confirm func() bool) error
	Rollback() error
}

TxFinalizer commits or rolls back the transaction a transactional subflow runs on.

Commit MUST:

  1. acquire the connector-facing operation lock and BLOCK for it - no cancellation, no timeout. A commit must never race a live statement;
  2. call confirm() WHILE HOLDING that lock and, if it returns false, ROLL BACK instead and return an error for which IsTxDowngraded reports true;
  3. release the pooled connection whatever happens.

Rollback MUST cancel any in-flight statement first, is best-effort about the operation lock, and must release the pooled connection whatever happens.

Both must be safe to call exactly once; the caller guarantees exactly-once through txScope.done.

type WorkItem 

type WorkItem struct {
	ID int `json:"id"`

	TaskID    string `json:"taskID"`
	SubFlowID int    `json:"subFlowId"`
	// contains filtered or unexported fields
}

WorkItem describes an item of work (event for a Task) that should be executed on Step

func NewWorkItem 

func NewWorkItem(id int, taskInst *TaskInst) *WorkItem

NewWorkItem constructs a new WorkItem for the specified TaskInst

type WorkingDataScope 

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

func NewWorkingDataScope 

func NewWorkingDataScope(parentScope data.Scope) *WorkingDataScope

NewWorkingDataScope

func (*WorkingDataScope) GetValue 

func (s *WorkingDataScope) GetValue(name string) (value interface{}, exists bool)

func (*WorkingDataScope) GetWorkingValue 

func (s *WorkingDataScope) GetWorkingValue(name string) (value interface{}, exists bool)

func (*WorkingDataScope) SetValue 

func (s *WorkingDataScope) SetValue(name string, value interface{}) error

func (*WorkingDataScope) SetWorkingValue 

func (s *WorkingDataScope) SetWorkingValue(name string, value interface{})

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.