Why Go
- Case Studies
  
  Common problems companies solve with Go
- Use Cases
  
  Stories about how and why companies use Go
- Security Policy
  
  How Go can help keep you secure by default
Learn
Docs
- Effective Go
  
  Tips for writing clear, performant, and idiomatic Go code
- Go User Manual
  
  A complete introduction to building software with Go
- Standard library
  
  Reference documentation for Go's standard library
- Release Notes
  
  Learn what's new in each Go release
Packages
Community
- Recorded Talks
  
  Videos from prior events
- Meetups
  
  Meet other local Go developers
- Conferences
  
  Learn and network with Go developers from around the world
- Go blog
  
  The Go project's official blog.
- Go project
  
  Get help and stay informed from Go
- Get connected

workflow

package module

Go to main page

Versions in this module

v0

v0.4.2

Dec 30, 2023

v0.4.1

Dec 22, 2023

v0.4.0

Dec 21, 2023

Changes in this version

+ var ErrCursorNotFound = errors.New("cursor not found")

+ var ErrRecordNotFound = errors.New("record not found")

+ var ErrRunIDNotFound = errors.New("run ID not found")

+ var ErrStatusProvidedNotConfigured = errors.New("status provided is not configured for workflow")

+ var ErrTimeoutNotFound = errors.New("timeout not found")

+ var ErrWorkflowInProgress = errors.New("current workflow still in progress - retry once complete")

+ var ErrWorkflowNotRunning = errors.New("trigger failed - workflow is not running")

+ var ErrWorkflowShutdown = errors.New("workflow has been shutdown")

+ func AwaitTimeoutInsert[Type any, Status StatusType](t *testing.T, w *Workflow[Type, Status], foreignID, runID string, ...)

+ func Marshal[T any](t *T) ([]byte, error)

+ func MermaidDiagram[Type any, Status StatusType](w *Workflow[Type, Status], path string, d MermaidDirection) error

+ func ParseTopic(topic string) (workflowName string, statusType int, err error)

+ func Require[Type any, Status StatusType](t *testing.T, w *Workflow[Type, Status], foreignID string, waitFor Status, ...)

+ func ToProto(r *WireRecord) *workflowpb.Record

+ func Topic(workflowName string, statusType int) string

+ func TriggerCallbackOn[Type any, Status StatusType, Payload any](t *testing.T, w *Workflow[Type, Status], foreignID, runID string, ...)

+ func Unmarshal[T any](b []byte, t *T) error

+ type API interface

+ Await func(ctx context.Context, foreignID, runID string, status Status, ...) (*Record[Type, Status], error)

+ Callback func(ctx context.Context, foreignID string, status Status, payload io.Reader) error

+ Run func(ctx context.Context)

+ ScheduleTrigger func(ctx context.Context, foreignID string, startingStatus Status, spec string, ...) error

+ Stop func()

+ Trigger func(ctx context.Context, foreignID string, startingStatus Status, ...) (runID string, err error)

+ type Ack func() error

+ type AwaitOption func(o *awaitOpts)

+ func WithPollingFrequency(d time.Duration) AwaitOption

+ type BuildOption func(w *buildOptions)

+ func WithClock(c clock.Clock) BuildOption

+ func WithDebugMode() BuildOption

+ type Builder struct

+ func NewBuilder[Type any, Status StatusType](name string) *Builder[Type, Status]

+ func (b *Builder[Type, Status]) AddCallback(from Status, fn CallbackFunc[Type, Status], to Status)

+ func (b *Builder[Type, Status]) AddConnector(name string, c Consumer, cf ConnectorFunc[Type, Status], ...)

+ func (b *Builder[Type, Status]) AddStep(from Status, c ConsumerFunc[Type, Status], to Status, opts ...StepOption)

+ func (b *Builder[Type, Status]) AddTimeout(from Status, timer TimerFunc[Type, Status], tf TimeoutFunc[Type, Status], ...)

+ func (b *Builder[Type, Status]) AddWorkflowConnector(cd WorkflowConnectionDetails, filter ConnectorFilter, from Status, ...)

+ func (b *Builder[Type, Status]) Build(eventStreamer EventStreamer, recordStore RecordStore, ...) *Workflow[Type, Status]

+ type CallbackFunc func(ctx context.Context, r *Record[Type, Status], reader io.Reader) (bool, error)

+ type ConnectorConsumerFunc func(ctx context.Context, r *Record[Type, Status], e *Event) (bool, error)

+ type ConnectorFilter func(ctx context.Context, e *Event) (foreignID string, err error)

+ type ConnectorFunc func(ctx context.Context, w *Workflow[Type, Status], e *Event) error

+ type ConnectorOption func(co *connectorOptions)

+ func WithConnectorErrBackOff(d time.Duration) ConnectorOption

+ func WithConnectorParallelCount(instances int) ConnectorOption

+ type Consumer interface

+ Close func() error

+ Recv func(ctx context.Context) (*Event, Ack, error)

+ type ConsumerFunc func(ctx context.Context, r *Record[Type, Status]) (bool, error)

+ func Not[Type any, Status StatusType](c ConsumerFunc[Type, Status]) ConsumerFunc[Type, Status]

+ type ConsumerOption func(*ConsumerOptions)

+ func WithConsumerPollFrequency(d time.Duration) ConsumerOption

+ func WithEventFilter(ef EventFilter) ConsumerOption

+ type ConsumerOptions struct

+ EventFilter EventFilter

+ PollFrequency time.Duration

+ type Cursor interface

+ Get func(ctx context.Context, name string) (string, error)

+ Set func(ctx context.Context, name, value string) error

+ type Event struct

+ CreatedAt time.Time

+ ForeignID int64

+ Headers map[Header]string

+ ID int64

+ Type int

+ type EventEmitter func(id int64) error

+ type EventFilter func(e *Event) bool

+ type EventStreamer interface

+ NewConsumer func(topic string, name string, opts ...ConsumerOption) Consumer

+ NewProducer func(topic string) Producer

+ type Header string

+ const HeaderRunID

+ const HeaderTopic

+ const HeaderWorkflowForeignID

+ const HeaderWorkflowName

+ type MermaidDirection string

+ const BottomToTopDirection

+ const LeftToRightDirection

+ const RightToLeftDirection

+ const TopToBottomDirection

+ const UnknownDirection

+ type MermaidFormat struct

+ Direction MermaidDirection

+ StartingPoints []string

+ TerminalPoints []string

+ Transitions []MermaidTransition

+ type MermaidTransition struct

+ From string

+ To string

+ type Producer interface

+ Close func() error

+ Send func(ctx context.Context, recordID int64, statusType int, headers map[Header]string) error

+ type Record struct

+ Object *Type

+ Status Status

+ type RecordStore interface

+ Latest func(ctx context.Context, workflowName, foreignID string) (*WireRecord, error)

+ Lookup func(ctx context.Context, id int64) (*WireRecord, error)

+ Store func(ctx context.Context, record *WireRecord, eventEmitter EventEmitter) error

+ type RoleScheduler interface

+ Await func(ctx context.Context, role string) (context.Context, context.CancelFunc, error)

+ type State string

+ const StateIdle

+ const StateRunning

+ const StateShutdown

+ const StateUnknown

+ type StatusType interface

+ String func() string

+ type StepOption func(so *stepOptions)

+ func WithParallelCount(instances int) StepOption

+ func WithStepErrBackOff(d time.Duration) StepOption

+ func WithStepLagAlert(d time.Duration) StepOption

+ func WithStepPollingFrequency(d time.Duration) StepOption

+ type TestingRecordStore interface

+ SetSnapshotOffset func(workflowName, foreignID, runID string, offset int)

+ SnapshotOffset func(workflowName, foreignID, runID string) int

+ Snapshots func(workflowName, foreignID, runID string) []*WireRecord

+ type Timeout struct

+ Completed bool

+ CreatedAt time.Time

+ ExpireAt time.Time

+ ForeignID string

+ ID int64

+ RunID string

+ Status int

+ WorkflowName string

+ type TimeoutFunc func(ctx context.Context, r *Record[Type, Status], now time.Time) (bool, error)

+ type TimeoutOption func(so *timeoutOptions)

+ func WithTimeoutErrBackOff(d time.Duration) TimeoutOption

+ func WithTimeoutLagAlert(d time.Duration) TimeoutOption

+ func WithTimeoutPollingFrequency(d time.Duration) TimeoutOption

+ type TimeoutStore interface

+ Cancel func(ctx context.Context, workflowName, foreignID, runID string, status int) error

+ Complete func(ctx context.Context, workflowName, foreignID, runID string, status int) error

+ Create func(ctx context.Context, workflowName, foreignID, runID string, status int, ...) error

+ List func(ctx context.Context, workflowName string) ([]Timeout, error)

+ ListValid func(ctx context.Context, workflowName string, status int, now time.Time) ([]Timeout, error)

+ type TimerFunc func(ctx context.Context, r *Record[Type, Status], now time.Time) (time.Time, error)

+ func DurationTimerFunc[Type any, Status StatusType](duration time.Duration) TimerFunc[Type, Status]

+ func TimeTimerFunc[Type any, Status StatusType](t time.Time) TimerFunc[Type, Status]

+ type TriggerOption func(o *triggerOpts[Type, Status])

+ func WithInitialValue[Type any, Status StatusType](t *Type) TriggerOption[Type, Status]

+ type WireRecord struct

+ CreatedAt time.Time

+ ForeignID string

+ ID int64

+ IsEnd bool

+ IsStart bool

+ Object []byte

+ RunID string

+ Status int

+ WorkflowName string

+ func UnmarshalRecord(b []byte) (*WireRecord, error)

+ func (r *WireRecord) ProtoMarshal() ([]byte, error)

+ type Workflow struct

+ Name string

+ func (w *Workflow[Type, Status]) Await(ctx context.Context, foreignID, runID string, status Status, ...) (*Record[Type, Status], error)

+ func (w *Workflow[Type, Status]) Callback(ctx context.Context, foreignID string, status Status, payload io.Reader) error

+ func (w *Workflow[Type, Status]) Run(ctx context.Context)

+ func (w *Workflow[Type, Status]) ScheduleTrigger(foreignID string, startingStatus Status, spec string, ...) error

+ func (w *Workflow[Type, Status]) States() map[string]State

+ func (w *Workflow[Type, Status]) Stop()

+ func (w *Workflow[Type, Status]) Trigger(ctx context.Context, foreignID string, startingStatus Status, ...) (runID string, err error)

+ type WorkflowConnectionDetails struct

+ Status int

+ Stream EventStreamer

+ WorkflowName string

v0.3.1

Dec 12, 2023

v0.3.0

Dec 7, 2023

v0.2.0

Dec 5, 2023

v0.1.3

Dec 4, 2023

v0.1.1

Dec 1, 2023

v0.1.0

Dec 1, 2023