interpreter

Documentation

Index

• func InterpreterImpl(ctx UnifiedContext, provider WorkflowProvider, ...) (*service.InterpreterWorkflowOutput, error)
• func RegisterActivityProvider(backendType service.BackendType, provider ActivityProvider)
• func StateDecide(ctx context.Context, backendType service.BackendType, ...) (*iwfidl.WorkflowStateDecideResponse, error)
• func StateStart(ctx context.Context, backendType service.BackendType, ...) (*iwfidl.WorkflowStateStartResponse, error)
• func WaitForDeciderTriggerType(provider WorkflowProvider, ctx UnifiedContext, ...)
• type ActivityOptions
• type ActivityProvider
• type Future
• type InterStateChannel
  • func NewInterStateChannel() *InterStateChannel
  • func RebuildInterStateChannel(refill map[string][]*iwfidl.EncodedObject) *InterStateChannel
  • func (i *InterStateChannel) HasData(channelName string) bool
  • func (i *InterStateChannel) ProcessPublishing(publishes []iwfidl.InterStateChannelPublishing)
  • func (i *InterStateChannel) ReadData(channelNames []string) map[string][]*iwfidl.EncodedObject
  • func (i *InterStateChannel) Retrieve(channelName string) *iwfidl.EncodedObject
• type PersistenceManager
  • func NewPersistenceManager(provider WorkflowProvider) *PersistenceManager
  • func RebuildPersistenceManager(provider WorkflowProvider, dolist []iwfidl.KeyValue, ...) *PersistenceManager
  • func (am *PersistenceManager) GetAllDataObjects() []iwfidl.KeyValue
  • func (am *PersistenceManager) GetAllSearchAttributes() []iwfidl.SearchAttribute
  • func (am *PersistenceManager) GetDataObjectsByKey(request service.GetDataObjectsQueryRequest) service.GetDataObjectsQueryResponse
  • func (am *PersistenceManager) LoadDataObjects(stateOptions *iwfidl.WorkflowStateOptions) []iwfidl.KeyValue
  • func (am *PersistenceManager) LoadSearchAttributes(stateOptions *iwfidl.WorkflowStateOptions) []iwfidl.SearchAttribute
  • func (am *PersistenceManager) ProcessUpsertDataObject(attributes []iwfidl.KeyValue) error
  • func (am *PersistenceManager) ProcessUpsertSearchAttribute(ctx UnifiedContext, attributes []iwfidl.SearchAttribute) error
• type ReceiveChannel
• type TimerProcessor
  • func NewTimerProcessor(ctx UnifiedContext, provider WorkflowProvider) *TimerProcessor
  • func (t *TimerProcessor) FinishProcessing(stateExeId string)
  • func (t *TimerProcessor) GetCurrentTimerInfos() map[string][]*service.TimerInfo
  • func (t *TimerProcessor) SkipTimer(stateExeId, timerId string, timerIdx int)
  • func (t *TimerProcessor) StartProcessing(stateExeId string, commands []iwfidl.TimerCommand)
  • func (t *TimerProcessor) WaitForTimerCompleted(ctx UnifiedContext, stateExeId string, timerIdx int, cancelWaiting *bool) bool
• type UnifiedContext
  • func NewUnifiedContext(ctx interface{}) UnifiedContext
• type UnifiedLogger
• type WorkflowExecution
• type WorkflowInfo
• type WorkflowProvider

Functions

func InterpreterImpl

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

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

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 ActivityOptions

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

type ActivityProvider

type ActivityProvider interface {
	GetLogger(ctx context.Context) UnifiedLogger
	NewApplicationError(message, errType string, details ...interface{}) 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

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) ReadData

func (i *InterStateChannel) ReadData(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) *PersistenceManager

func RebuildPersistenceManager

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)
	ReceiveAsync(valuePtr interface{}) (ok bool)
}

type TimerProcessor

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

func NewTimerProcessor

func NewTimerProcessor(ctx UnifiedContext, provider WorkflowProvider) *TimerProcessor

func (*TimerProcessor) FinishProcessing

func (t *TimerProcessor) FinishProcessing(stateExeId string)

func (*TimerProcessor) GetCurrentTimerInfos

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

func (*TimerProcessor) SkipTimer

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

func (*TimerProcessor) StartProcessing

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

func (*TimerProcessor) WaitForTimerCompleted

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

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(message, errType string, details ...interface{}) error
	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
	GetBackendType() service.BackendType
	GetLogger(ctx UnifiedContext) UnifiedLogger
}