interpreter

package
v1.3.2 Latest Latest
Warning

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

 Go to latest Published: Feb 13, 2023 License: MIT Imports: 10 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func InterpreterImpl 

func InterpreterImpl(ctx UnifiedContext, provider WorkflowProvider, input service.InterpreterWorkflowInput) (*service.InterpreterWorkflowOutput, error)

func NewGlobalVersionProvider added in v1.3.0 

func NewGlobalVersionProvider(workflowProvider WorkflowProvider) *globalVersioner

func RegisterActivityProvider 

func RegisterActivityProvider(backendType service.BackendType, provider ActivityProvider)

func StateDecide 

func StateDecide(ctx context.Context, backendType service.BackendType, input service.StateDecideActivityInput) (*iwfidl.WorkflowStateDecideResponse, error)

func StateStart 

func StateStart(ctx context.Context, backendType service.BackendType, input service.StateStartActivityInput) (*iwfidl.WorkflowStateStartResponse, error)

func WaitForDeciderTriggerType added in v1.2.0 

func WaitForDeciderTriggerType(
	provider WorkflowProvider,
	ctx UnifiedContext,
	commandReq iwfidl.CommandRequest,
	completedTimerCmds map[int]bool,
	completedSignalCmds map[int]*iwfidl.EncodedObject,
	completedInterStateChannelCmds map[int]*iwfidl.EncodedObject,
)

Types

type ActivityInfo added in v1.2.2 

type ActivityInfo struct {
	ScheduledTime time.Time // Time of activity scheduled by a workflow
	Attempt       int32     // Attempt starts from 1, and increased by 1 for every retry if retry policy is specified.
}

type ActivityOptions 

type ActivityOptions struct {
	StartToCloseTimeout time.Duration
	RetryPolicy         *iwfidl.RetryPolicy
}

type ActivityProvider 

type ActivityProvider interface {
	GetLogger(ctx context.Context) UnifiedLogger
	NewApplicationError(errType string, details interface{}) error
	GetActivityInfo(ctx context.Context) ActivityInfo
}

type ContinueAsNewer added in v1.3.0 

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

func NewContinueAsNewer added in v1.3.0 

func NewContinueAsNewer(
	interStateChannel *InterStateChannel, signalReceiver *SignalReceiver, stateExecutionCounter *StateExecutionCounter, persistenceManager *PersistenceManager,
) *ContinueAsNewer

func (*ContinueAsNewer) AddPendingStateExecution added in v1.3.0 

func (c *ContinueAsNewer) AddPendingStateExecution(
	stateExecutionId string,
	completedTimerCommands map[int]bool, completedSignalCommands, completedInterStateChannelCommands map[int]*iwfidl.EncodedObject,
	timerCommands []iwfidl.TimerCommand, signalCommands []iwfidl.SignalCommand, interStateChannelCommands []iwfidl.InterStateChannelCommand,
)

func (*ContinueAsNewer) DeletePendingStateExecution added in v1.3.0 

func (c *ContinueAsNewer) DeletePendingStateExecution(stateExecutionId string)

func (*ContinueAsNewer) SetQueryHandlersForContinueAsNew added in v1.3.0 

func (c *ContinueAsNewer) SetQueryHandlersForContinueAsNew(ctx UnifiedContext, provider WorkflowProvider) error

type Future 

type Future interface {
	Get(ctx UnifiedContext, valuePtr interface{}) error
	IsReady() bool
}

type InterStateChannel 

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

func NewInterStateChannel 

func NewInterStateChannel() *InterStateChannel

func RebuildInterStateChannel added in v1.2.0 

func RebuildInterStateChannel(refill map[string][]*iwfidl.EncodedObject) *InterStateChannel

func (*InterStateChannel) HasData 

func (i *InterStateChannel) HasData(channelName string) bool

func (*InterStateChannel) ProcessPublishing 

func (i *InterStateChannel) ProcessPublishing(publishes []iwfidl.InterStateChannelPublishing)

func (*InterStateChannel) ReadReceived added in v1.3.0 

func (i *InterStateChannel) ReadReceived(channelNames []string) map[string][]*iwfidl.EncodedObject

func (*InterStateChannel) Retrieve 

func (i *InterStateChannel) Retrieve(channelName string) *iwfidl.EncodedObject

type PersistenceManager 

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

func NewPersistenceManager 

func NewPersistenceManager(provider WorkflowProvider, initSearchAttributes []iwfidl.SearchAttribute) *PersistenceManager

func RebuildPersistenceManager added in v1.2.0 

func RebuildPersistenceManager(provider WorkflowProvider,
	dolist []iwfidl.KeyValue, salist []iwfidl.SearchAttribute,
) *PersistenceManager

func (*PersistenceManager) GetAllDataObjects 

func (am *PersistenceManager) GetAllDataObjects() []iwfidl.KeyValue

func (*PersistenceManager) GetAllSearchAttributes 

func (am *PersistenceManager) GetAllSearchAttributes() []iwfidl.SearchAttribute

func (*PersistenceManager) GetDataObjectsByKey 

func (am *PersistenceManager) GetDataObjectsByKey(request service.GetDataObjectsQueryRequest) service.GetDataObjectsQueryResponse

func (*PersistenceManager) LoadDataObjects 

func (am *PersistenceManager) LoadDataObjects(stateOptions *iwfidl.WorkflowStateOptions) []iwfidl.KeyValue

func (*PersistenceManager) LoadSearchAttributes 

func (am *PersistenceManager) LoadSearchAttributes(stateOptions *iwfidl.WorkflowStateOptions) []iwfidl.SearchAttribute

func (*PersistenceManager) ProcessUpsertDataObject 

func (am *PersistenceManager) ProcessUpsertDataObject(attributes []iwfidl.KeyValue) error

func (*PersistenceManager) ProcessUpsertSearchAttribute 

func (am *PersistenceManager) ProcessUpsertSearchAttribute(ctx UnifiedContext, attributes []iwfidl.SearchAttribute) error

type ReceiveChannel 

type ReceiveChannel interface {
	Receive(ctx UnifiedContext, valuePtr interface{}) (more bool) // TODO: check with Temporal about the API semantics -- Cadence says the return is "ok" but Temporal says it's "more"
	ReceiveAsync(valuePtr interface{}) (ok bool)
}

type SignalReceiver added in v1.3.0 

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

func NewSignalReceiver added in v1.3.0 

func NewSignalReceiver(ctx UnifiedContext, provider WorkflowProvider) *SignalReceiver

func (*SignalReceiver) DrainedAllSignals added in v1.3.0 

func (sr *SignalReceiver) DrainedAllSignals(ctx UnifiedContext) error

DrainedAllSignals will wait for all signals are processed before a safe continueAsNew

func (*SignalReceiver) HasSignal added in v1.3.0 

func (sr *SignalReceiver) HasSignal(channelName string) bool

func (*SignalReceiver) ReadReceived added in v1.3.0 

func (sr *SignalReceiver) ReadReceived(channelNames []string) map[string][]*iwfidl.EncodedObject

func (*SignalReceiver) Retrieve added in v1.3.0 

func (sr *SignalReceiver) Retrieve(channelName string) *iwfidl.EncodedObject

type StateExecutionCounter added in v1.3.0 

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

func NewStateExecutionCounter added in v1.3.0 

func NewStateExecutionCounter(ctx UnifiedContext, provider WorkflowProvider, config service.WorkflowConfig) *StateExecutionCounter

func RebuildStateExecutionManager added in v1.3.0 

func RebuildStateExecutionManager(ctx UnifiedContext, provider WorkflowProvider,
	executedStateIdCount map[string]int, pendingStateIdCount map[string]int, totalPendingStateExeCount int,
) *StateExecutionCounter

func (*StateExecutionCounter) CreateNextExecutionId added in v1.3.0 

func (e *StateExecutionCounter) CreateNextExecutionId(stateId string) string

func (*StateExecutionCounter) Dump added in v1.3.0 

func (e *StateExecutionCounter) Dump() service.StateExecutionCounterInfo

func (*StateExecutionCounter) GetTotalPendingStateExecutions added in v1.3.0 

func (e *StateExecutionCounter) GetTotalPendingStateExecutions() int

func (*StateExecutionCounter) MarkStateExecutionCompleted added in v1.3.0 

func (e *StateExecutionCounter) MarkStateExecutionCompleted(state iwfidl.StateMovement) error

func (*StateExecutionCounter) MarkStateExecutionsPending added in v1.3.0 

func (e *StateExecutionCounter) MarkStateExecutionsPending(states []iwfidl.StateMovement) error

type TimerProcessor added in v1.2.0 

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

func NewTimerProcessor added in v1.2.0 

func NewTimerProcessor(ctx UnifiedContext, provider WorkflowProvider) *TimerProcessor

func (*TimerProcessor) FinishProcessing added in v1.2.0 

func (t *TimerProcessor) FinishProcessing(stateExeId string)

func (*TimerProcessor) GetCurrentTimerInfos added in v1.2.0 

func (t *TimerProcessor) GetCurrentTimerInfos() map[string][]*service.TimerInfo

func (*TimerProcessor) SkipTimer added in v1.2.0 

func (t *TimerProcessor) SkipTimer(stateExeId, timerId string, timerIdx int)

func (*TimerProcessor) StartProcessing added in v1.2.0 

func (t *TimerProcessor) StartProcessing(stateExeId string, commands []iwfidl.TimerCommand)

func (*TimerProcessor) WaitForTimerCompleted added in v1.2.0 

func (t *TimerProcessor) WaitForTimerCompleted(ctx UnifiedContext, stateExeId string, timerIdx int, cancelWaiting *bool) bool

WaitForTimerCompleted waits for timer completed(fired or skipped), return false if the waiting is canceled by cancelWaiting bool pointer

type UnifiedContext 

type UnifiedContext interface {
	GetContext() interface{}
}

func NewUnifiedContext 

func NewUnifiedContext(ctx interface{}) UnifiedContext

type UnifiedLogger added in v1.2.0 

type UnifiedLogger interface {
	Debug(msg string, keyvals ...interface{})
	Info(msg string, keyvals ...interface{})
	Warn(msg string, keyvals ...interface{})
	Error(msg string, keyvals ...interface{})
}

type WorkflowExecution 

type WorkflowExecution struct {
	ID    string
	RunID string
}

WorkflowExecution details.

type WorkflowInfo 

type WorkflowInfo struct {
	WorkflowExecution WorkflowExecution
	WorkflowStartTime time.Time
}

WorkflowInfo information about currently executing workflow

type WorkflowProvider 

type WorkflowProvider interface {
	NewApplicationError(errType string, details interface{}) error
	IsApplicationError(err error) bool
	GetWorkflowInfo(ctx UnifiedContext) WorkflowInfo
	UpsertSearchAttributes(ctx UnifiedContext, attributes map[string]interface{}) error
	SetQueryHandler(ctx UnifiedContext, queryType string, handler interface{}) error
	ExtendContextWithValue(parent UnifiedContext, key string, val interface{}) UnifiedContext
	GoNamed(ctx UnifiedContext, name string, f func(ctx UnifiedContext))
	Await(ctx UnifiedContext, condition func() bool) error
	WithActivityOptions(ctx UnifiedContext, options ActivityOptions) UnifiedContext
	ExecuteActivity(ctx UnifiedContext, activity interface{}, args ...interface{}) (future Future)
	Now(ctx UnifiedContext) time.Time
	Sleep(ctx UnifiedContext, d time.Duration) (err error)
	NewTimer(ctx UnifiedContext, d time.Duration) Future
	GetSignalChannel(ctx UnifiedContext, signalName string) (receiveChannel ReceiveChannel)
	GetContextValue(ctx UnifiedContext, key string) interface{}
	GetVersion(ctx UnifiedContext, changeID string, minSupported, maxSupported int) int
	GetUnhandledSignalNames(ctx UnifiedContext) []string
	GetBackendType() service.BackendType
	GetLogger(ctx UnifiedContext) UnifiedLogger
}

Source Files

View all Source files

Directories

Path Synopsis

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.