internal

Documentation

Index

• func CommandName(cmd any) (string, error)
• func InitCommand(name string) (any, error)
• type ActivityInfo
• type Cancel
• type CancelExternalWorkflow
• type CancelWorkflow
• type CompleteWorkflow
• type Context
  • func (ctx Context) IsEmpty() bool
• type ContinueAsNew
• type DestroyWorkflow
• type ExecuteActivity
  • func (cmd ExecuteActivity) ActivityParams(env bindings.WorkflowEnvironment, payloads *commonpb.Payloads) bindings.ExecuteActivityParams
• type ExecuteChildWorkflow
  • func (cmd ExecuteChildWorkflow) WorkflowParams(env bindings.WorkflowEnvironment, payloads *commonpb.Payloads) bindings.ExecuteWorkflowParams
• type ExecuteLocalActivity
  • func (cmd ExecuteLocalActivity) LocalActivityParams(env bindings.WorkflowEnvironment, fn any, payloads *commonpb.Payloads) bindings.ExecuteLocalActivityParams
• type ExecuteLocalActivityOptions
• type GetChildWorkflowExecution
• type GetStackTrace
• type GetVersion
• type GetWorkerInfo
• type InvokeActivity
• type InvokeLocalActivity
• type InvokeQuery
• type InvokeSignal
• type Message
  • func (msg *Message) IsCommand() bool
  • func (msg *Message) Reset()
• type NewTimer
  • func (cmd NewTimer) ToDuration() time.Duration
• type Panic
• type SideEffect
• type SignalExternalWorkflow
• type StartWorkflow
• type UpsertWorkflowSearchAttributes
• type WorkerInfo
• type WorkflowInfo

Functions

func CommandName

func CommandName(cmd any) (string, error)

CommandName returns command name (only for the commands sent to the worker)

func InitCommand

func InitCommand(name string) (any, error)

InitCommand reads command from binary payload

Types

type ActivityInfo

type ActivityInfo struct {
	// Name describes public activity name.
	Name string `json:"name"`
}

ActivityInfo describes single worker activity.

type Cancel

type Cancel struct {
	// CommandIDs to be canceled.
	CommandIDs []uint64 `json:"ids"`
}

Cancel one or multiple internal promises (activities, local activities, timers, child workflows).

type CancelExternalWorkflow

type CancelExternalWorkflow struct {
	Namespace  string `json:"namespace"`
	WorkflowID string `json:"workflowID"`
	RunID      string `json:"runID"`
}

CancelExternalWorkflow canceller external workflow.

type CancelWorkflow

type CancelWorkflow struct {
	// RunID workflow run id.
	RunID string `json:"runId"`
}

CancelWorkflow asks worker to gracefully stop workflow, if possible (signal).

type CompleteWorkflow

type CompleteWorkflow struct{}

CompleteWorkflow sent by worker to complete workflow. Might include additional error as part of the payload.

type Context

type Context struct {
	// TaskQueue associates message batch with the specific task queue in underlying worker.
	TaskQueue string `json:"taskQueue,omitempty"`

	// TickTime associated current or historical time with message batch.
	TickTime string `json:"tickTime,omitempty"`

	// Replay indicates that current message batch is historical.
	Replay bool `json:"replay,omitempty"`
}

Context provides worker information about currently. Context can be empty for server level commands.

func (Context) IsEmpty

func (ctx Context) IsEmpty() bool

IsEmpty only check if task queue set.

type ContinueAsNew

type ContinueAsNew struct {
	// Result defines workflow execution result.
	Name string `json:"name"`

	// Options for continued as new workflow.
	Options struct {
		TaskQueueName       string
		WorkflowRunTimeout  time.Duration
		WorkflowTaskTimeout time.Duration
	} `json:"options"`
}

ContinueAsNew restarts workflow with new running instance.

type DestroyWorkflow

type DestroyWorkflow struct {
	// RunID workflow run id.
	RunID string `json:"runId"`
}

DestroyWorkflow asks worker to offload workflow from memory.

type ExecuteActivity

type ExecuteActivity struct {
	// Name defines activity name.
	Name string `json:"name"`
	// Options to run activity.
	Options bindings.ExecuteActivityOptions `json:"options,omitempty"`
}

ExecuteActivity command by workflow worker.

func (ExecuteActivity) ActivityParams

func (cmd ExecuteActivity) ActivityParams(env bindings.WorkflowEnvironment, payloads *commonpb.Payloads) bindings.ExecuteActivityParams

ActivityParams maps activity command to activity params.

type ExecuteChildWorkflow

type ExecuteChildWorkflow struct {
	// Name defines workflow name.
	Name string `json:"name"`
	// Options to run activity.
	Options bindings.WorkflowOptions `json:"options,omitempty"`
}

ExecuteChildWorkflow executes child workflow.

func (ExecuteChildWorkflow) WorkflowParams

func (cmd ExecuteChildWorkflow) WorkflowParams(env bindings.WorkflowEnvironment, payloads *commonpb.Payloads) bindings.ExecuteWorkflowParams

WorkflowParams maps workflow command to workflow params.

type ExecuteLocalActivity

type ExecuteLocalActivity struct {
	// Name defines activity name.
	Name string `json:"name"`
	// Options to run activity.
	Options ExecuteLocalActivityOptions `json:"options,omitempty"`
}

ExecuteLocalActivity command by workflow worker.

func (ExecuteLocalActivity) LocalActivityParams

func (cmd ExecuteLocalActivity) LocalActivityParams(env bindings.WorkflowEnvironment, fn any, payloads *commonpb.Payloads) bindings.ExecuteLocalActivityParams

LocalActivityParams maps activity command to activity params.

type ExecuteLocalActivityOptions

type ExecuteLocalActivityOptions struct {
	ScheduleToCloseTimeout time.Duration
	StartToCloseTimeout    time.Duration
	RetryPolicy            *commonpb.RetryPolicy
}

ExecuteLocalActivityOptions .. since we use proto everywhere we need to convert Activity options (proto) to non-proto LA options

type GetChildWorkflowExecution

type GetChildWorkflowExecution struct {
	// ID of child workflow command.
	ID uint64 `json:"id"`
}

GetChildWorkflowExecution returns the WorkflowID and RunId of child workflow.

type GetStackTrace

type GetStackTrace struct {
	// RunID workflow run id.
	RunID string `json:"runId"`
}

GetStackTrace asks worker to offload workflow from memory.

type GetVersion

type GetVersion struct {
	ChangeID     string `json:"changeID"`
	MinSupported int    `json:"minSupported"`
	MaxSupported int    `json:"maxSupported"`
}

GetVersion requests version marker.

type GetWorkerInfo

type GetWorkerInfo struct {
	RRVersion string `json:"rr_version"`
}

GetWorkerInfo reads worker information.

type InvokeActivity

type InvokeActivity struct {
	// Name defines activity name.
	Name string `json:"name"`

	// Info contains execution context.
	Info activity.Info `json:"info"`

	// HeartbeatDetails indicates that the payload also contains last heartbeat details.
	HeartbeatDetails int `json:"heartbeatDetails,omitempty"`
}

InvokeActivity invokes activity.

type InvokeLocalActivity

type InvokeLocalActivity struct {
	// Name defines activity name.
	Name string `json:"name"`

	// Info contains execution context.
	Info activity.Info `json:"info"`
}

InvokeLocalActivity invokes local activity.

type InvokeQuery

type InvokeQuery struct {
	// RunID workflow run id.
	RunID string `json:"runId"`
	// Name of the query.
	Name string `json:"name"`
}

InvokeQuery invokes query with a set of arguments.

type InvokeSignal

type InvokeSignal struct {
	// RunID workflow run id.
	RunID string `json:"runId"`

	// Name of the signal.
	Name string `json:"name"`
}

InvokeSignal invokes signal with a set of arguments.

type Message

type Message struct {
	// ID contains ID of the command, response or error.
	ID uint64 `json:"id"`

	// Command of the message in unmarshalled form. Pointer.
	Command any `json:"command,omitempty"`

	// Failure associated with command id.
	Failure *failure.Failure `json:"failure,omitempty"`

	// Payloads contains message specific payloads in binary format.
	Payloads *commonpb.Payloads `json:"payloads,omitempty"`

	// Header
	Header *commonpb.Header `json:"header,omitempty"`
}

Message used to exchange the send commands and receive responses from underlying workers.

func (*Message) IsCommand

func (msg *Message) IsCommand() bool

IsCommand returns true if message carries request.

func (*Message) Reset

func (msg *Message) Reset()

type NewTimer

type NewTimer struct {
	// Milliseconds defines timer duration.
	Milliseconds int `json:"ms"`
}

NewTimer starts new timer.

func (NewTimer) ToDuration

func (cmd NewTimer) ToDuration() time.Duration

ToDuration converts timer command to time.Duration.

type Panic

type Panic struct {
	// Message to include into the error.
	Message string `json:"message"`
}

Panic triggers panic in workflow process.

type SideEffect

type SideEffect struct{}

SideEffect to be recorded into the history.

type SignalExternalWorkflow

type SignalExternalWorkflow struct {
	Namespace         string `json:"namespace"`
	WorkflowID        string `json:"workflowID"`
	RunID             string `json:"runID"`
	Signal            string `json:"signal"`
	ChildWorkflowOnly bool   `json:"childWorkflowOnly"`
}

SignalExternalWorkflow sends signal to external workflow.

type StartWorkflow

type StartWorkflow struct {
	// Info to define workflow context.
	Info *workflow.Info `json:"info"`

	// LastCompletion contains offset of last completion results.
	LastCompletion int `json:"lastCompletion,omitempty"`
}

StartWorkflow sends worker command to start workflow.

type UpsertWorkflowSearchAttributes

type UpsertWorkflowSearchAttributes struct {
	SearchAttributes map[string]any `json:"searchAttributes"`
}

UpsertWorkflowSearchAttributes allows to upsert search attributes

type WorkerInfo

type WorkerInfo struct {
	// TaskQueue assigned to the worker.
	TaskQueue string `json:"taskQueue"`

	// Options describe worker options.
	Options worker.Options `json:"options,omitempty"`

	// Workflows provided by the worker.
	Workflows []WorkflowInfo

	// Activities provided by the worker.
	Activities []ActivityInfo
}

WorkerInfo lists available task queues, workflows and activities.

type WorkflowInfo

type WorkflowInfo struct {
	// Name of the workflow.
	Name string `json:"name"`

	// Queries pre-defined for the workflow type.
	Queries []string `json:"queries"`

	// Signals pre-defined for the workflow type.
	Signals []string `json:"signals"`
}

WorkflowInfo describes single worker workflow.