workflow

Documentation

Index

• Variables
• func CanRetry(err error) bool
• func ContinueAsNew(ctx Context, args ...any) error
• func Go(ctx Context, f func(ctx Context))
• func Logger(ctx Context) *slog.Logger
• func NewError(err error) error
• func NewPermanentError(err error) error
• func Now(ctx Context) time.Time
• func Replaying(ctx Context) bool
• func Select(ctx Context, cases ...SelectCase)
• func Sleep(ctx Context, d time.Duration) error
• func WithCancel(parent Context) (ctx Context, cancel CancelFunc)
• type Activity
• type ActivityOptions
• type CancelFunc
• type Channel
  • func NewBufferedChannel[T any](size int) Channel[T]
  • func NewChannel[T any]() Channel[T]
  • func NewSignalChannel[T any](ctx Context, name string) Channel[T]
• type Context
  • func NewDisconnectedContext(ctx Context) Context
  • func WithValue(parent Context, key, val interface{}) Context
• type ContextPropagator
• type Error
• type Future
  • func CreateSubWorkflowInstance[TResult any](ctx Context, options SubWorkflowOptions, workflow Workflow, args ...any) Future[TResult]
  • func ExecuteActivity[TResult any](ctx Context, options ActivityOptions, activity Activity, args ...any) Future[TResult]
  • func ScheduleTimer(ctx Context, delay time.Duration) Future[any]
  • func SideEffect[TResult any](ctx Context, f func(ctx Context) TResult) Future[TResult]
  • func SignalWorkflow[T any](ctx Context, instanceID string, name string, arg T) Future[any]
  • func WithRetries[T any](ctx Context, retryOptions RetryOptions, ...) Future[T]
• type Instance
  • func WorkflowInstance(ctx Context) *Instance
• type Metadata
• type PanicError
• type RetryOptions
• type SelectCase
  • func Await[T any](f Future[T], handler func(Context, Future[T])) SelectCase
  • func Default(handler func(Context)) SelectCase
  • func Receive[T any](c Channel[T], handler func(ctx Context, v T, ok bool)) SelectCase
  • func Send[T any](c Channel[T], value *T, handler func(ctx Context)) SelectCase
• type Span
• type SubWorkflowOptions
• type WaitGroup
  • func NewWaitGroup() WaitGroup
• type Workflow
• type WorkflowTracer
  • func Tracer(ctx Context) *WorkflowTracer
  • func (wt *WorkflowTracer) Start(ctx Context, name string, opts ...trace.SpanStartOption) (Context, Span)

Variables

var (
	DefaultSubWorkflowRetryOptions = RetryOptions{

		MaxAttempts: 1,
	}

	DefaultSubWorkflowOptions = SubWorkflowOptions{
		RetryOptions: DefaultSubWorkflowRetryOptions,
	}
)

var Canceled = sync.Canceled

var DefaultActivityOptions = ActivityOptions{
	RetryOptions: DefaultRetryOptions,
}

var DefaultRetryOptions = RetryOptions{
	MaxAttempts:        3,
	BackoffCoefficient: 1,
}

Functions

func CanRetry

func CanRetry(err error) bool

CanRetry returns true if the given error is retryable

func ContinueAsNew

func ContinueAsNew(ctx Context, args ...any) error

ContinueAsNew restarts the current workflow with the given arguments.

func Go

func Go(ctx Context, f func(ctx Context))

Go spawns a workflow "goroutine".

func Logger

func Logger(ctx Context) *slog.Logger

Logger returns the logger for the current workflow.

func NewError

func NewError(err error) error

NewError wraps the given error into a workflow error which will be automatically retried

func NewPermanentError

func NewPermanentError(err error) error

NewPermanentError wraps the given error into a workflow error which will not be automatically retried

func Now

func Now(ctx Context) time.Time

Now returns the current time.

func Replaying

func Replaying(ctx Context) bool

Replaying returns true if the current workflow execution is replaying or not.

func Select

func Select(ctx Context, cases ...SelectCase)

Select is the workflow-save equivalent of the select statement.

func Sleep

func Sleep(ctx Context, d time.Duration) error

Sleep sleeps for the given duration.

func WithCancel

func WithCancel(parent Context) (ctx Context, cancel CancelFunc)

WithCancel returns a copy of parent with a new Done channel. The returned context's Done channel is closed when the returned cancel function is called or when the parent context's Done channel is closed, whichever happens first.

Canceling this context releases resources associated with it, so code should call cancel as soon as the operations running in this Context complete.

Types

type Activity

type Activity = interface{}

type ActivityOptions

type ActivityOptions struct {
	// RetryOptions defines how to retry the activity in case of failure.
	RetryOptions RetryOptions
}

type CancelFunc

type CancelFunc = sync.CancelFunc

type Channel

type Channel[T any] interface {
	// Send sends a value to the channel. If the channel is closed, this will panic.
	Send(ctx Context, v T)

	// SendNonblocking sends a value to the channel. This call is non-blocking and will return whether the
	// value could be sent. If the channel is closed, this will panic
	SendNonblocking(v T) (ok bool)

	// Receive receives a value from the channel.
	Receive(ctx Context) (v T, ok bool)

	// ReceiveNonBlocking tries to receives a value from the channel. This call is non-blocking and will return
	// whether the value could be returned.
	ReceiveNonBlocking() (v T, ok bool)

	// Close closes the channel. This will cause all future send operations to panic.
	Close()
}

func NewBufferedChannel

func NewBufferedChannel[T any](size int) Channel[T]

NewBufferedChannel creates a new buffered channel with the given size.

func NewChannel

func NewChannel[T any]() Channel[T]

NewChannel creates a new channel.

func NewSignalChannel

func NewSignalChannel[T any](ctx Context, name string) Channel[T]

NewSignalChannel returns a new signal channel.

type Context

type Context = sync.Context

func NewDisconnectedContext

func NewDisconnectedContext(ctx Context) Context

NewDisconnectedContext creates a new context that is disconnected from any parent context.

func WithValue

func WithValue(parent Context, key, val interface{}) Context

WithValue returns a copy of parent in which the value associated with key is val.

Use context Values only for request-scoped data that transits processes and APIs, not for passing optional parameters to functions.

The provided key must be comparable and should not be of type string or any other built-in type to avoid collisions between packages using context. Users of WithValue should define their own types for keys. To avoid allocating when assigning to an interface{}, context keys often have concrete type struct{}. Alternatively, exported context key variables' static type should be a pointer or interface.

type ContextPropagator

type ContextPropagator interface {
	// Inject injects values from context into metadata
	Inject(context.Context, *Metadata) error

	// Extract extracts values from metadata into context
	Extract(context.Context, *Metadata) (context.Context, error)

	// InjectFromWorkflow injects values from the workflow context into metadata
	InjectFromWorkflow(Context, *Metadata) error

	// ExtractToWorkflow extracts values from metadata into a workflow context
	ExtractToWorkflow(Context, *Metadata) (Context, error)
}

type Error

type Error = workflowerrors.Error

type Future

type Future[T any] interface {
	// Get returns the value if set, blocks otherwise
	Get(ctx Context) (T, error)
}

func CreateSubWorkflowInstance

func CreateSubWorkflowInstance[TResult any](ctx Context, options SubWorkflowOptions, workflow Workflow, args ...any) Future[TResult]

CreateSubWorkflowInstance creates a new sub-workflow instance of the given workflow.

func ExecuteActivity

func ExecuteActivity[TResult any](ctx Context, options ActivityOptions, activity Activity, args ...any) Future[TResult]

ExecuteActivity schedules the given activity to be executed

func ScheduleTimer

func ScheduleTimer(ctx Context, delay time.Duration) Future[any]

ScheduleTimer schedules a timer to fire after the given delay.

func SideEffect

func SideEffect[TResult any](ctx Context, f func(ctx Context) TResult) Future[TResult]

SideEffect executes the given function and returns a future that will be resolved with the result of the function.

In contrast to Activities, SideEffects are executed inline with the workflow code. They should only be used for short and inexpensive operations. For longer operations, consider using an Activity.

func SignalWorkflow

func SignalWorkflow[T any](ctx Context, instanceID string, name string, arg T) Future[any]

SignalWorkflow sends a signal to another running workflow instance.

func WithRetries

func WithRetries[T any](ctx Context, retryOptions RetryOptions, fn func(ctx Context, attempt int) Future[T]) Future[T]

WithRetries executes the given function with retries.

type Instance

type Instance = core.WorkflowInstance

Instance represents a workflow instance.

func WorkflowInstance

func WorkflowInstance(ctx Context) *Instance

WorkflowInstance returns the current workflow instance.

type Metadata

type Metadata = metadata.WorkflowMetadata

Metadata represents the metadata of a workflow instance.

type PanicError

type PanicError = workflowerrors.PanicError

type RetryOptions

type RetryOptions struct {
	// Maximum number of times to retry
	MaxAttempts int

	// Time to wait before first retry
	FirstRetryInterval time.Duration

	// Maximum delay for any individual retry attempt
	MaxRetryInterval time.Duration

	// Coeffecient for calculation the next retry delay
	BackoffCoefficient float64

	// Timeout after which retries are aborted
	RetryTimeout time.Duration
}

type SelectCase

type SelectCase = sync.SelectCase

func Await

func Await[T any](f Future[T], handler func(Context, Future[T])) SelectCase

Await calls the provided handler when the given future is ready.

func Default

func Default(handler func(Context)) SelectCase

Default calls the provided handler if none of the other cases match.

func Receive

func Receive[T any](c Channel[T], handler func(ctx Context, v T, ok bool)) SelectCase

Receive calls the provided handler if the given channel can receive a value. The handler receives the received value, and the ok flag indicating whether the value was received or the channel was closed.

func Send

func Send[T any](c Channel[T], value *T, handler func(ctx Context)) SelectCase

Send calls the provided handler if the given value can be sent to the channel.

type Span

type Span interface {
	End()
}

type SubWorkflowOptions

type SubWorkflowOptions struct {
	InstanceID string

	RetryOptions RetryOptions
}

type WaitGroup

type WaitGroup = sync.WaitGroup

func NewWaitGroup

func NewWaitGroup() WaitGroup

NewWaitGroup creates a new WaitGroup instance.

type Workflow

type Workflow = interface{}

type WorkflowTracer

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

func Tracer

func Tracer(ctx Context) *WorkflowTracer

Tracer creates a the workflow tracer.

func (*WorkflowTracer) Start

func (wt *WorkflowTracer) Start(ctx Context, name string, opts ...trace.SpanStartOption) (Context, Span)

Start starts a new span.