service

Documentation

Index

• Constants
• Variables
• type BackendType
• type BasicInfo
• type ContinueAsNewDumpResponse
• type ContinueAsNewInput
• type DebugDumpResponse
• type ExecuteRpcSignalRequest
• type FailWorkflowSignalRequest
• type GetCurrentTimerInfosQueryResponse
• type GetDataObjectsQueryRequest
• type GetDataObjectsQueryResponse
• type InternalTimerStatus
• type InterpreterWorkflowInput
• type InterpreterWorkflowOutput
• type PrepareRpcQueryRequest
• type PrepareRpcQueryResponse
• type SkipTimerSignalRequest
• type StaleSkipTimerSignal
• type StateDecideActivityInput
• type StateExecutionCompletedCommands
• type StateExecutionCounterInfo
• type StateExecutionResumeInfo
• type StateExecutionStatus
• type StateStartActivityInput
• type TimerInfo
  • func ValidateTimerSkipRequest(stateExeTimerInfos map[string][]*TimerInfo, stateExeId, timerId string, ...) (*TimerInfo, bool)

Constants

const (
	DefaultContinueAsNewPageSizeInBytes = 1024 * 1024

	// HttpStatusCodeWorkerApiError is a special deprecated code for this because I can't find an official one for this case
	HttpStatusCodeWorkerApiError = 420

	TaskQueue = "Interpreter_DEFAULT"

	StateStartApi        = "/api/v1/workflowState/start"
	StateDecideApi       = "/api/v1/workflowState/decide"
	WorkflowWorkerRpcApi = "/api/v1/workflowWorker/rpc"

	GetDataObjectsWorkflowQueryType      = "GetDataObjects"
	GetSearchAttributesWorkflowQueryType = "GetSearchAttributes"
	GetCurrentTimerInfosQueryType        = "GetCurrentTimerInfos"
	ContinueAsNewDumpQueryType           = "ContinueAsNewDump"
	DebugDumpQueryType                   = "DebugNewDump"
	PrepareRpcQueryType                  = "PrepareRpcQueryType"

	SearchAttributeGlobalVersion     = "IwfGlobalWorkflowVersion"
	SearchAttributeExecutingStateIds = "IwfExecutingStateIds"
	SearchAttributeIwfWorkflowType   = "IwfWorkflowType"

	BackendTypeCadence  BackendType = "cadence"
	BackendTypeTemporal BackendType = "temporal"

	IwfSystemSignalPrefix         = "__IwfSystem_"
	SkipTimerSignalChannelName    = IwfSystemSignalPrefix + "SkipTimerChannel"
	FailWorkflowSignalChannelName = IwfSystemSignalPrefix + "FailWorkflowChannel"
	UpdateConfigSignalChannelName = IwfSystemSignalPrefix + "UpdateWorkflowConfig"
	ExecuteRpcSignalChannelName   = IwfSystemSignalPrefix + "ExecuteRpc"
)

const (
	GracefulCompletingWorkflowStateId = "_SYS_GRACEFUL_COMPLETING_WORKFLOW"
	ForceCompletingWorkflowStateId    = "_SYS_FORCE_COMPLETING_WORKFLOW"
	ForceFailingWorkflowStateId       = "_SYS_FORCE_FAILING_WORKFLOW"
	DeadEndWorkflowStateId            = "_SYS_DEAD_END"
)

Variables

var ValidClosingWorkflowStateId = map[string]bool{
	GracefulCompletingWorkflowStateId: true,
	ForceCompletingWorkflowStateId:    true,
	ForceFailingWorkflowStateId:       true,
	DeadEndWorkflowStateId:            true,
}

var ValidIwfSystemSignalNames = map[string]bool{
	SkipTimerSignalChannelName:    true,
	FailWorkflowSignalChannelName: true,
	UpdateConfigSignalChannelName: true,
	ExecuteRpcSignalChannelName:   true,
}

Types

type BackendType

type BackendType string

type BasicInfo

type BasicInfo struct {
	IwfWorkflowType string `json:"iwfWorkflowType,omitempty"`

	IwfWorkerUrl string `json:"iwfWorkerUrl,omitempty"`
}

type ContinueAsNewDumpResponse

type ContinueAsNewDumpResponse struct {
	StatesToStartFromBeginning []iwfidl.StateMovement              // StatesToStartFromBeginning means they haven't started in the previous run
	StateExecutionsToResume    map[string]StateExecutionResumeInfo // stateExeId to StateExecutionResumeInfo
	InterStateChannelReceived  map[string][]*iwfidl.EncodedObject
	SignalsReceived            map[string][]*iwfidl.EncodedObject
	StateExecutionCounterInfo  StateExecutionCounterInfo
	StateOutputs               []iwfidl.StateCompletionOutput
	StaleSkipTimerSignals      []StaleSkipTimerSignal

	DataObjects      []iwfidl.KeyValue
	SearchAttributes []iwfidl.SearchAttribute
}

type ContinueAsNewInput

type ContinueAsNewInput struct {
	PreviousInternalRunId string `json:"previousInternalRunId"` // for loading from previous run
}

type DebugDumpResponse

type DebugDumpResponse struct {
	Config iwfidl.WorkflowConfig
}

type ExecuteRpcSignalRequest

type ExecuteRpcSignalRequest struct {
	RpcInput                    *iwfidl.EncodedObject
	RpcOutput                   *iwfidl.EncodedObject
	UpsertDataObjects           []iwfidl.KeyValue
	UpsertSearchAttributes      []iwfidl.SearchAttribute
	StateDecision               *iwfidl.StateDecision
	RecordEvents                []iwfidl.KeyValue
	InterStateChannelPublishing []iwfidl.InterStateChannelPublishing
}

type FailWorkflowSignalRequest

type FailWorkflowSignalRequest struct {
	Reason string
}

type GetCurrentTimerInfosQueryResponse

type GetCurrentTimerInfosQueryResponse struct {
	StateExecutionCurrentTimerInfos map[string][]*TimerInfo // key is stateExecutionId
}

type GetDataObjectsQueryRequest

type GetDataObjectsQueryRequest struct {
	Keys []string
}

type GetDataObjectsQueryResponse

type GetDataObjectsQueryResponse struct {
	DataObjects []iwfidl.KeyValue
}

type InternalTimerStatus

type InternalTimerStatus string

const (
	TimerPending InternalTimerStatus = "Pending"
	TimerFired   InternalTimerStatus = "Fired"
	TimerSkipped InternalTimerStatus = "Skipped"
)

type InterpreterWorkflowInput

type InterpreterWorkflowInput struct {
	IwfWorkflowType string `json:"iwfWorkflowType,omitempty"`

	IwfWorkerUrl string `json:"iwfWorkerUrl,omitempty"`

	StartStateId *string `json:"startStateId,omitempty"`

	StateInput iwfidl.EncodedObject `json:"stateInput,omitempty"`

	StateOptions iwfidl.WorkflowStateOptions `json:"stateOptions,omitempty"`

	InitSearchAttributes []iwfidl.SearchAttribute `json:"initSearchAttributes,omitempty"`

	Config iwfidl.WorkflowConfig `json:"config,omitempty"`

	// IsResumeFromContinueAsNew indicate this is input for continueAsNew
	// when true, will ignore StartStateId, StateInput, StateOptions, InitSearchAttributes
	IsResumeFromContinueAsNew bool `json:"isResumeFromContinueAsNew"`

	ContinueAsNewInput ContinueAsNewInput `json:"continueAsNewInput"`
}

type InterpreterWorkflowOutput

type InterpreterWorkflowOutput struct {
	StateCompletionOutputs []iwfidl.StateCompletionOutput `json:"stateCompletionOutputs,omitempty"`
}

type PrepareRpcQueryRequest

type PrepareRpcQueryRequest struct {
	DataObjectsLoadingPolicy      *iwfidl.PersistenceLoadingPolicy
	SearchAttributesLoadingPolicy *iwfidl.PersistenceLoadingPolicy
}

type PrepareRpcQueryResponse

type PrepareRpcQueryResponse struct {
	DataObjects              []iwfidl.KeyValue
	SearchAttributes         []iwfidl.SearchAttribute
	WorkflowRunId            string
	WorkflowStartedTimestamp int64
	IwfWorkflowType          string
	IwfWorkerUrl             string
}

type SkipTimerSignalRequest

type SkipTimerSignalRequest struct {
	StateExecutionId string
	CommandId        string
	CommandIndex     int
}

type StaleSkipTimerSignal

type StaleSkipTimerSignal struct {
	StateExecutionId  string
	TimerCommandId    string
	TimerCommandIndex int
}

type StateDecideActivityInput

type StateDecideActivityInput struct {
	IwfWorkerUrl string
	Request      iwfidl.WorkflowStateDecideRequest
}

type StateExecutionCompletedCommands

type StateExecutionCompletedCommands struct {
	CompletedTimerCommands             map[int]InternalTimerStatus   `json:"completedTimerCommands"`
	CompletedSignalCommands            map[int]*iwfidl.EncodedObject `json:"completedSignalCommands"`
	CompletedInterStateChannelCommands map[int]*iwfidl.EncodedObject `json:"completedInterStateChannelCommands"`
}

type StateExecutionCounterInfo

type StateExecutionCounterInfo struct {
	StateIdStartedCount            map[string]int // for stateExecutionId
	StateIdCurrentlyExecutingCount map[string]int // for sys search attribute ExecutingStateIds
	TotalCurrentlyExecutingCount   int            // for "dead end"
}

type StateExecutionResumeInfo

type StateExecutionResumeInfo struct {
	StateExecutionId                string                          `json:"stateExecutionId"`
	State                           iwfidl.StateMovement            `json:"state"`
	StateExecutionCompletedCommands StateExecutionCompletedCommands `json:"stateExecutionCompletedCommands"`
	CommandRequest                  iwfidl.CommandRequest           `json:"commandRequest"`
	StateExecutionLocals            []iwfidl.KeyValue               `json:"stateExecutionLocals"`
}

type StateExecutionStatus

type StateExecutionStatus string

const CompletedStateExecutionStatus StateExecutionStatus = "Completed" // this will process as normal

const FailureStateExecutionStatus StateExecutionStatus = "Failure"

const WaitingCommandsStateExecutionStatus StateExecutionStatus = "WaitingCommands" // this will put the state into a special pending queue for continueAsNew from waiting command

type StateStartActivityInput

type StateStartActivityInput struct {
	IwfWorkerUrl string
	Request      iwfidl.WorkflowStateStartRequest
}

type TimerInfo

type TimerInfo struct {
	CommandId                  string
	FiringUnixTimestampSeconds int64
	Status                     InternalTimerStatus
}

func ValidateTimerSkipRequest

func ValidateTimerSkipRequest(stateExeTimerInfos map[string][]*TimerInfo, stateExeId, timerId string, timerIdx int) (*TimerInfo, bool)

ValidateTimerSkipRequest validates if the skip timer request is valid return true if it's valid, along with the timer pointer use timerIdx if timerId is not empty