workflow

package
v1.1.0 Latest Latest
Warning

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

 Go to latest Published: Aug 26, 2026 License: Apache-2.0 Imports: 11 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	StatusRunning        = api.RUNTIME_STATUS_RUNNING
	StatusCompleted      = api.RUNTIME_STATUS_COMPLETED
	StatusContinuedAsNew = api.RUNTIME_STATUS_CONTINUED_AS_NEW
	StatusFailed         = api.RUNTIME_STATUS_FAILED
	StatusCanceled       = api.RUNTIME_STATUS_CANCELED
	StatusTerminated     = api.RUNTIME_STATUS_TERMINATED
	StatusPending        = api.RUNTIME_STATUS_PENDING
	StatusSuspended      = api.RUNTIME_STATUS_SUSPENDED
	StatusStalled        = api.RUNTIME_STATUS_STALLED
)

Variables

This section is empty.

Functions

func WithHistoryPropagation 

func WithHistoryPropagation(opt PropagationOption) task.HistoryPropagation

WithHistoryPropagation creates an option that configures history propagation for a child workflow or activity call. It can be passed to both CallActivity and CallChildWorkflow. Pass one of PropagateOwnHistory() or PropagateLineage().

Usage: 

ctx.CallChildWorkflow("child",
    workflow.WithChildWorkflowInput(input),
    workflow.WithHistoryPropagation(workflow.PropagateLineage()),
)

ctx.CallActivity("activity",
    workflow.WithActivityInput(input),
    workflow.WithHistoryPropagation(workflow.PropagateOwnHistory()),
)

func WorkflowMetadataIsComplete 

func WorkflowMetadataIsComplete(o *WorkflowMetadata) bool

func WorkflowMetadataIsRunning 

func WorkflowMetadataIsRunning(o *WorkflowMetadata) bool

Types

type Activity 

type Activity func(ActivityContext) (any, error)

Activity is the functional interface for activity implementations.

type ActivityContext 

type ActivityContext task.ActivityContext

ActivityContext is the context parameter type for activity implementations.

type CallActivityOption 

type CallActivityOption task.CallActivityOption

func WithActivityAppID 

func WithActivityAppID(targetAppID string) CallActivityOption

func WithActivityAppNamespace 

func WithActivityAppNamespace(namespace string) CallActivityOption

WithActivityAppNamespace specifies the Dapr namespace that hosts the target activity. Must be combined with WithActivityAppID. See task.WithActivityAppNamespace for full semantics.

func WithActivityInput 

func WithActivityInput(input any) CallActivityOption

WithActivityInput configures an input for an activity invocation. The specified input must be JSON serializable.

func WithActivityRetryPolicy 

func WithActivityRetryPolicy(policy *RetryPolicy) CallActivityOption

func WithRawActivityInput 

func WithRawActivityInput(input *wrapperspb.StringValue) CallActivityOption

WithRawActivityInput configures a raw input for an activity invocation.

type ChildWorkflowOption 

type ChildWorkflowOption task.ChildWorkflowOption

ChildWorkflowOption is a functional option type for the CallChildWorkflow workflow method.

func WithChildWorkflowAppID 

func WithChildWorkflowAppID(appID string) ChildWorkflowOption

WithChildWorkflowAppID is a functional option type for the CallChildWorkflow workflow method that specifies the app ID of the target activity.

func WithChildWorkflowAppNamespace 

func WithChildWorkflowAppNamespace(namespace string) ChildWorkflowOption

WithChildWorkflowAppNamespace specifies the Dapr namespace that hosts the target child workflow. Must be combined with WithChildWorkflowAppID. See task.WithChildWorkflowAppNamespace for full semantics.

func WithChildWorkflowInput 

func WithChildWorkflowInput(input any) ChildWorkflowOption

WithChildWorkflowInput is a functional option type for the CallChildWorkflow workflow method that takes an input value and marshals it to JSON.

func WithChildWorkflowInstanceID 

func WithChildWorkflowInstanceID(instanceID string) ChildWorkflowOption

WithChildWorkflowInstanceID is a functional option type for the CallChildWorkflow workflow method that specifies the instance ID of the child-workflow.

func WithChildWorkflowRetryPolicy 

func WithChildWorkflowRetryPolicy(policy *RetryPolicy) ChildWorkflowOption

func WithRawChildWorkflowInput 

func WithRawChildWorkflowInput(input *wrapperspb.StringValue) ChildWorkflowOption

WithRawChildWorkflowInput is a functional option type for the CallChildWorkflow workflow method that takes a raw input value.

type Client 

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

func NewClient 

func NewClient(cc grpc.ClientConnInterface) *Client

NewClient creates a client that can be used to manage worfklows.

func NewClientWithLogger 

func NewClientWithLogger(cc grpc.ClientConnInterface, logger backend.Logger) *Client

func (*Client) FetchWorkflowMetadata 

func (c *Client) FetchWorkflowMetadata(ctx context.Context, id string, opts ...FetchWorkflowMetadataOptions) (*WorkflowMetadata, error)

FetchWorkflowMetadata fetches metadata for the specified workflow from the configured task hub.

api.ErrInstanceNotFound is returned when the specified workflow doesn't exist.

func (*Client) GetInstanceHistory 

func (c *Client) GetInstanceHistory(ctx context.Context, id string, opts ...GetInstanceHistoryOptions) (*GetInstanceHistoryResponse, error)

func (*Client) ListInstanceIDs 

func (c *Client) ListInstanceIDs(ctx context.Context, opts ...ListInstanceIDsOptions) (*ListInstanceIDsResponse, error)

func (*Client) PurgeWorkflowState 

func (c *Client) PurgeWorkflowState(ctx context.Context, id string, opts ...PurgeOptions) error

PurgeWorkflowState deletes the state of the specified workflow instance.

[api.api.ErrInstanceNotFound] is returned if the specified workflow instance doesn't exist.

func (*Client) RaiseEvent 

func (c *Client) RaiseEvent(ctx context.Context, id, eventName string, opts ...RaiseEventOptions) error

RaiseEvent sends an asynchronous event notification to a waiting workflow.

func (*Client) ResumeWorkflow 

func (c *Client) ResumeWorkflow(ctx context.Context, id, reason string, opts ...ResumeOptions) error

ResumeWorkflow resumes a workflow instance that was previously suspended.

func (*Client) ScheduleWorkflow 

func (c *Client) ScheduleWorkflow(ctx context.Context, workflow string, opts ...NewWorkflowOptions) (string, error)

ScheduleWorkflow schedules a new workflow instance with a specified set of options for execution.

func (*Client) StartWorker 

func (c *Client) StartWorker(ctx context.Context, r *Registry) error

StartWorker starts the workflow runtime to process workflows.

func (*Client) SuspendWorkflow 

func (c *Client) SuspendWorkflow(ctx context.Context, id, reason string, opts ...SuspendOptions) error

SuspendWorkflow suspends an workflow instance, halting processing of its events until a "resume" operation resumes it.

Note that suspended workflows are still considered to be "running" even though they will not process events.

func (*Client) TerminateWorkflow 

func (c *Client) TerminateWorkflow(ctx context.Context, id string, opts ...TerminateOptions) error

TerminateWorkflow terminates a running workflow by causing it to stop receiving new events and putting it directly into the TERMINATED state.

func (*Client) WaitForWorkflowCompletion 

func (c *Client) WaitForWorkflowCompletion(ctx context.Context, id string, opts ...FetchWorkflowMetadataOptions) (*WorkflowMetadata, error)

WaitForWorkflowCompletion waits for an workflow to complete and returns an backend.WorkflowMetadata object that contains metadata about the completed instance.

api.ErrInstanceNotFound is returned when the specified workflow doesn't exist.

func (*Client) WaitForWorkflowStart 

func (c *Client) WaitForWorkflowStart(ctx context.Context, id string, opts ...FetchWorkflowMetadataOptions) (*WorkflowMetadata, error)

WaitForWorkflowStart waits for an workflow to start running and returns an backend.WorkflowMetadata object that contains metadata about the started instance.

api.ErrInstanceNotFound is returned when the specified workflow doesn't exist.

type ContinueAsNewOption 

type ContinueAsNewOption task.ContinueAsNewOption

ContinueAsNewOption is a functional option type for the ContinueAsNew workflow method.

func WithKeepUnprocessedEvents 

func WithKeepUnprocessedEvents() ContinueAsNewOption

WithKeepUnprocessedEvents returns a ContinueAsNewOptions struct that instructs the runtime to carry forward any unprocessed external events to the new instance.

type CreateTimerOption 

type CreateTimerOption task.CreateTimerOption

func WithTimerName 

func WithTimerName(name string) CreateTimerOption

type DetachedWorkflowOptions 

type DetachedWorkflowOptions task.DetachedWorkflowOptions

DetachedWorkflowOptions is a functional option type for the ScheduleNewWorkflow workflow method.

func WithDetachedWorkflowAppID 

func WithDetachedWorkflowAppID(appID string) DetachedWorkflowOptions

WithDetachedWorkflowAppID specifies the app ID hosting the detached workflow. When set, the action carries a routing envelope so the dispatcher delivers the new instance to the target app.

func WithDetachedWorkflowAppNamespace 

func WithDetachedWorkflowAppNamespace(namespace string) DetachedWorkflowOptions

WithDetachedWorkflowAppNamespace specifies the Dapr namespace hosting the detached workflow. Must be combined with WithDetachedWorkflowAppID.

func WithDetachedWorkflowInput 

func WithDetachedWorkflowInput(input any) DetachedWorkflowOptions

WithDetachedWorkflowInput sets the JSON-marshalled input for the detached workflow.

func WithDetachedWorkflowInstanceID 

func WithDetachedWorkflowInstanceID(instanceID string) DetachedWorkflowOptions

WithDetachedWorkflowInstanceID sets the instance ID of the detached workflow. When omitted, ScheduleNewWorkflow generates a deterministic ID of the form "<callerInstanceID>-<n>". Passing an empty string is rejected as an error: callers either set a non-empty ID or omit the option entirely to opt into the default.

func WithDetachedWorkflowStartTime 

func WithDetachedWorkflowStartTime(startTime time.Time) DetachedWorkflowOptions

WithDetachedWorkflowStartTime defers the start of the detached workflow until the given time.

func WithRawDetachedWorkflowInput 

func WithRawDetachedWorkflowInput(input *wrapperspb.StringValue) DetachedWorkflowOptions

WithRawDetachedWorkflowInput sets a pre-marshalled raw input on the detached workflow.

type FetchWorkflowMetadataOptions 

type FetchWorkflowMetadataOptions api.FetchWorkflowMetadataOptions

func WithFetchAppID 

func WithFetchAppID(appID string) FetchWorkflowMetadataOptions

WithFetchAppID targets the metadata fetch at the workflow instance owned by the app with the given app ID rather than the local app.

func WithFetchPayloads 

func WithFetchPayloads(fetchPayloads bool) FetchWorkflowMetadataOptions

WithFetchPayloads configures whether to load workflow inputs, outputs, and custom status values, which could be large.

type GetInstanceHistoryOptions 

type GetInstanceHistoryOptions api.GetInstanceHistoryOptions

type GetInstanceHistoryResponse 

type GetInstanceHistoryResponse protos.GetInstanceHistoryResponse

type ListInstanceIDsOptions 

type ListInstanceIDsOptions api.ListInstanceIDsOptions

func WithListInstanceIDsContinuationToken 

func WithListInstanceIDsContinuationToken(token string) ListInstanceIDsOptions

func WithListInstanceIDsPageSize 

func WithListInstanceIDsPageSize(pageSize uint32) ListInstanceIDsOptions

type ListInstanceIDsResponse 

type ListInstanceIDsResponse protos.ListInstanceIDsResponse

type NewWorkflowOptions 

type NewWorkflowOptions api.NewWorkflowOptions

func WithAppID 

func WithAppID(appID string) NewWorkflowOptions

WithAppID targets the new workflow at the app with the given app ID rather than the local app.

func WithInput 

func WithInput(input any) NewWorkflowOptions

WithInput configures an input for the workflow. The specified input must be serializable.

func WithInstanceID 

func WithInstanceID(id string) NewWorkflowOptions

WithInstanceID configures an explicit workflow instance ID. If not specified, a random UUID value will be used for the workflow instance ID.

func WithRawInput 

func WithRawInput(rawInput *wrapperspb.StringValue) NewWorkflowOptions

WithRawInput configures an input for the workflow. The specified input must be a string.

func WithStartTime 

func WithStartTime(startTime time.Time) NewWorkflowOptions

WithStartTime configures a start time at which the workflow should start running. Note that the actual start time could be later than the specified start time if the task hub is under load or if the app is not running at the specified start time.

type PropagatedHistory 

type PropagatedHistory = api.PropagatedHistory

PropagatedHistory is the history propagated from a parent workflow to a child workflow or activity.

type PropagationOption 

type PropagationOption = api.PropagationOption

PropagationOption configures how history is propagated to child workflows & activities.

func PropagateLineage 

func PropagateLineage() PropagationOption

PropagateLineage configures propagation to include the caller's own history events AND the full ancestor chain. Any propagated history this workflow received from its own parent is forwarded to the child, preserving the full chain of custody.

func PropagateOwnHistory 

func PropagateOwnHistory() PropagationOption

PropagateOwnHistory configures propagation to include the caller's own history events (activities, child workflows, timers, etc.). The child only sees events from the immediate parent, not the ancestor chain.

type PurgeOptions 

type PurgeOptions api.PurgeOptions

func WithForcePurge 

func WithForcePurge(force bool) PurgeOptions

func WithPurgeAppID 

func WithPurgeAppID(appID string) PurgeOptions

WithPurgeAppID targets the purge at the workflow instance owned by the app with the given app ID rather than the local app. The purge is delegated to the target app, which honours the caller's recursive flag.

func WithRecursivePurge 

func WithRecursivePurge(recursive bool) PurgeOptions

WithRecursivePurge configures whether to purge all child-workflows created by the target workflow.

type RaiseEventOptions 

type RaiseEventOptions api.RaiseEventOptions

func WithEventPayload 

func WithEventPayload(data any) RaiseEventOptions

WithEventPayload configures an event payload. The specified payload must be serializable.

func WithExternalDeliveryID added in v1.1.0 

func WithExternalDeliveryID(deliveryID string) RaiseEventOptions

WithExternalDeliveryID identifies an at-least-once external delivery.

func WithRaiseEventAppID 

func WithRaiseEventAppID(appID string) RaiseEventOptions

WithRaiseEventAppID targets the event at the workflow instance owned by the app with the given app ID rather than the local app.

func WithRawEventData 

func WithRawEventData(data *wrapperspb.StringValue) RaiseEventOptions

WithRawEventData configures an event payload that is a raw, unprocessed string (e.g. JSON data).

type Registry 

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

Registry contains maps of names to corresponding workflow and activity functions.

func NewRegistry 

func NewRegistry() *Registry

NewRegistry returns a new Registry struct.

func (*Registry) AddActivity 

func (r *Registry) AddActivity(a Activity) error

AddActivity adds an activity function to the registry. The name of the activity function is determined using reflection.

func (*Registry) AddActivityN 

func (r *Registry) AddActivityN(name string, a Activity) error

AddActivityN adds an activity function to the registry with a specified name.

func (*Registry) AddVersionedWorkflow 

func (r *Registry) AddVersionedWorkflow(canonicalName string, isLatest bool, w Workflow) error

AddVersionedWorkflow adds a versioned workflow function to the registry with a specified name.

func (*Registry) AddVersionedWorkflowN 

func (r *Registry) AddVersionedWorkflowN(canonicalName string, name string, isLatest bool, w Workflow) error

AddVersionedWorkflowN adds a versioned workflow function to the registry with a specified name.

func (*Registry) AddWorkflow 

func (r *Registry) AddWorkflow(w Workflow) error

AddWorkflow adds a workflow function to the registry. The name of the workflow function is determined using reflection.

func (*Registry) AddWorkflowN 

func (r *Registry) AddWorkflowN(name string, w Workflow) error

AddWorkflowN adds a workflow function to the registry with a specified name.

type ResumeOptions 

type ResumeOptions api.ResumeOptions

func WithResumeAppID 

func WithResumeAppID(appID string) ResumeOptions

WithResumeAppID targets the resumption at the workflow instance owned by the app with the given app ID rather than the local app.

type RetryPolicy 

type RetryPolicy struct {
	// Max number of attempts to try the activity call, first execution inclusive
	MaxAttempts int
	// Timespan to wait for the first retry
	InitialRetryInterval time.Duration
	// Used to determine rate of increase of back-off
	BackoffCoefficient float64
	// Max timespan to wait for a retry
	MaxRetryInterval time.Duration
	// Total timeout across all the retries performed
	RetryTimeout time.Duration
	// Optional function to control if retries should proceed
	Handle func(error) bool
}

type SuspendOptions 

type SuspendOptions api.SuspendOptions

func WithSuspendAppID 

func WithSuspendAppID(appID string) SuspendOptions

WithSuspendAppID targets the suspension at the workflow instance owned by the app with the given app ID rather than the local app.

type Task 

type Task task.Task

Task is an interface for asynchronous durable tasks. A task is conceptually similar to a future.

type TerminateOptions 

type TerminateOptions api.TerminateOptions

func WithOutput 

func WithOutput(data any) TerminateOptions

WithOutput configures an output for the terminated workflow. The specified output must be serializable.

func WithRawOutput 

func WithRawOutput(data *wrapperspb.StringValue) TerminateOptions

WithRawOutput configures a raw, unprocessed output (i.e. pre-serialized) for the terminated workflow.

func WithRecursiveTerminate 

func WithRecursiveTerminate(recursive bool) TerminateOptions

WithRecursiveTerminate configures whether to terminate all child-workflows created by the target workflow.

func WithTerminateAppID 

func WithTerminateAppID(appID string) TerminateOptions

WithTerminateAppID targets the termination at the workflow instance owned by the app with the given app ID rather than the local app.

type Workflow 

type Workflow func(ctx *WorkflowContext) (any, error)

Workflow is the functional interface for workflow functions.

type WorkflowContext 

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

WorkflowContext is the parameter type for workflow functions.

func (*WorkflowContext) CallActivity 

func (w *WorkflowContext) CallActivity(activity any, opts ...CallActivityOption) Task

CallActivity schedules an asynchronous invocation of an activity function. The [activity] parameter can be either the name of an activity as a string or can be a pointer to the function that implements the activity, in which case the name is obtained via reflection.

func (*WorkflowContext) CallChildWorkflow 

func (w *WorkflowContext) CallChildWorkflow(workflow any, opts ...ChildWorkflowOption) Task

func (*WorkflowContext) ContinueAsNew 

func (w *WorkflowContext) ContinueAsNew(newInput any, options ...ContinueAsNewOption)

func (*WorkflowContext) CreateTimer 

func (w *WorkflowContext) CreateTimer(delay time.Duration, opts ...CreateTimerOption) Task

CreateTimer schedules a durable timer that expires after the specified delay.

func (*WorkflowContext) CurrentTimeUTC 

func (w *WorkflowContext) CurrentTimeUTC() time.Time

func (*WorkflowContext) GetInput 

func (w *WorkflowContext) GetInput(v any) error

GetInput unmarshals the serialized workflow input and stores it in [v].

func (*WorkflowContext) GetPropagatedHistory 

func (w *WorkflowContext) GetPropagatedHistory() *PropagatedHistory

GetPropagatedHistory returns the propagated history from a parent workflow, or nil if no history was propagated.

func (*WorkflowContext) ID 

func (w *WorkflowContext) ID() string

func (*WorkflowContext) IsPatched 

func (w *WorkflowContext) IsPatched(patchName string) bool

func (*WorkflowContext) IsReplaying 

func (w *WorkflowContext) IsReplaying() bool

func (*WorkflowContext) Name 

func (w *WorkflowContext) Name() string

func (*WorkflowContext) ScheduleNewWorkflow 

func (w *WorkflowContext) ScheduleNewWorkflow(workflow any, opts ...DetachedWorkflowOptions) (string, error)

ScheduleNewWorkflow schedules a new, fully decoupled workflow instance from this workflow. Returns the new instance ID synchronously. The spawned workflow has no parent linkage: completion and failure do not flow back to the caller. If WithDetachedWorkflowInstanceID is not provided, a deterministic default ID is generated.

func (*WorkflowContext) SetCustomStatus 

func (w *WorkflowContext) SetCustomStatus(cs string)

func (*WorkflowContext) WaitForExternalEvent 

func (w *WorkflowContext) WaitForExternalEvent(eventName string, timeout time.Duration) Task

WaitForExternalEvent creates a task that is completed only after an event named [eventName] is received by this workflow or when the specified timeout expires.

The [timeout] parameter can be used to define a timeout for receiving the event. If the timeout expires before the named event is received, the task will be completed and will return a timeout error value [ErrTaskCanceled] when awaited. Otherwise, the awaited task will return the deserialized payload of the received event. A Duration value of zero returns a canceled task if the event isn't already available in the history. Use a negative Duration to wait indefinitely for the event to be received.

Workflows can wait for the same event name multiple times, so waiting for multiple events with the same name is allowed. Each event received by an workflow will complete just one task returned by this method.

Note that event names are case-insensitive.

type WorkflowMetadata 

type WorkflowMetadata struct {
	InstanceId       string
	Name             string
	RuntimeStatus    protos.OrchestrationStatus
	CreatedAt        *timestamppb.Timestamp
	LastUpdatedAt    *timestamppb.Timestamp
	Input            *wrapperspb.StringValue
	Output           *wrapperspb.StringValue
	CustomStatus     *wrapperspb.StringValue
	FailureDetails   *protos.TaskFailureDetails
	CompletedAt      *timestamppb.Timestamp
	ParentInstanceId string
	Version          *wrapperspb.StringValue
	ParentAppId      *wrapperspb.StringValue
	StartedAt        *timestamppb.Timestamp
}

WorkflowMetadata is the SDK representation of workflow metadata. It mirrors the application fields returned by the service without embedding protobuf runtime state, so values can safely implement fmt.Stringer.

func (WorkflowMetadata) String 

func (w WorkflowMetadata) String() string

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.