riverworkflow

package module

Go to main page

Versions in this module

v0

v0.3.0

Jul 13, 2026

Changes in this version

+ func AwaitSignal[T any](ctx Context, name string) T

+ func RegisterActivity[TArgs river.JobArgs, TOut any](workers *river.Workers, fn func(context.Context, TArgs) (TOut, error))

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

+ func Signal[TTx any, T any](ctx context.Context, client *river.Client[TTx], workflowID, name string, ...) error

+ type CommandRecord struct

+ Args json.RawMessage

+ CommandID int

+ Kind string

+ Output json.RawMessage

+ Role string

+ State string

type EngineOpts

+ Store Store

+ type HistoryStore interface

+ Checkpoint func(ctx context.Context, id string, gen int, commands []CommandRecord, ...) error

+ LoadHistory func(ctx context.Context, id string, gen int) ([]CommandRecord, []SignalRecord, error)

+ PruneGenerationsBefore func(ctx context.Context, id string, gen int) error

+ PruneWorkflow func(ctx context.Context, id string) error

+ type MissingOutputError struct

+ Kind string

+ func (e *MissingOutputError) Error() string

+ type SelectCase struct

+ func OnFuture[T any](f *Future[T], handler func(T, error)) SelectCase

+ func OnSignal[T any](name string, handler func(T)) SelectCase

+ func OnTimeout(d time.Duration, handler func()) SelectCase

+ type SignalRecord struct

+ Name string

+ Payload json.RawMessage

+ SignalID int64

type Status

+ const StatusFailed

+ type Store interface

+ LoadState func(ctx context.Context, id string) (WorkflowState, error)

+ RecordCancelled func(ctx context.Context, id string) error

+ RecordCompleted func(ctx context.Context, id string, result json.RawMessage) error

+ RecordFailed func(ctx context.Context, id, failure string) error

+ RecordStarted func(ctx context.Context, id, kind string, gen int, runJobID int64) error

+ RunJobID func(ctx context.Context, id string) (runJobID int64, ok bool, err error)

+ type Wake struct

+ ID string

+ RunJobID int64

+ type WakeStore interface

+ ClearWake func(ctx context.Context, id string) error

+ MarkWake func(ctx context.Context, id string) error

+ PendingWakes func(ctx context.Context, limit int) ([]Wake, error)

+ WakeEvents func() <-chan string

+ type WorkflowState struct

+ Failure string

+ Found bool

+ Gen int

+ Result json.RawMessage

+ Status Status

+ func (s WorkflowState) Terminal() bool

v0.2.0

Jul 10, 2026

Changes in this version

+ func ContinueAsNew[I any](ctx Context, in I) error

type Future

+ func ChildWorkflow[I any, O any](ctx Context, kind string, in I) *Future[O]

v0.1.0

Jul 10, 2026

Changes in this version

+ var ErrWorkflowFailed = errors.New("riverworkflow: workflow did not complete successfully")

+ func Cancel[TTx any](ctx context.Context, client *river.Client[TTx], workflowID string) error

+ func Now(ctx Context) time.Time

+ func Random(ctx Context) int64

+ func RegisterWorkflow[I, O any](reg *Registry, kind string, fn func(ctx Context, in I) (O, error))

+ func Result[TTx any, O any](ctx context.Context, client *river.Client[TTx], workflowID string) (O, bool, error)

+ func SideEffect[T any](ctx Context, fn func() T) T

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

+ func Start[TTx any, I any](ctx context.Context, client *river.Client[TTx], kind string, in I, ...) (string, error)

+ type ActivityError struct

+ Kind string

+ State rivertype.JobState

+ func (e *ActivityError) Error() string

+ type ActivityOpts struct

+ InsertOpts *river.InsertOpts

+ type Context struct

+ func (c Context) Context() context.Context

+ type Engine struct

+ func NewEngine[TTx any](reg *Registry, opts *EngineOpts) *Engine[TTx]

+ func (e *Engine[TTx]) Install(config *river.Config, workers *river.Workers)

+ func (e *Engine[TTx]) Start(ctx context.Context, client *river.Client[TTx])

+ func (e *Engine[TTx]) Stop()

+ type EngineOpts struct

+ ReconcileInterval time.Duration

+ SnoozeInterval time.Duration

+ type Future struct

+ func Execute[TArgs river.JobArgs, TOut any](ctx Context, args TArgs, opts *ActivityOpts) *Future[TOut]

+ func (f *Future[TOut]) Get() (TOut, error)

+ type Registry struct

+ func NewRegistry() *Registry

+ type RunWorker struct

+ func (w *RunWorker[TTx]) Work(ctx context.Context, job *river.Job[runArgs]) error

+ type StartOpts struct

+ ID string

+ type Status string

+ const StatusCancelled

+ const StatusCompleted

+ const StatusDiscarded

+ const StatusRunning

+ func GetStatus[TTx any](ctx context.Context, client *river.Client[TTx], workflowID string) (Status, error)

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL