Versions in this module Expand all Collapse all v0 v0.3.0 Nov 10, 2026 Changes in this version + const BatchJobDeliveryType + const CallbackDeliveryType + const ChainNodeDeliveryType + const ContinuationDeliveryType + const DirectDeliveryType + const ProtocolSchemaVersion + var ErrContinuationProtocol = errors.New("workflow continuation protocol rejected") + var ErrHandlerNotRegistered = errors.New("workflow handler is not registered") + var ErrNotFound = errors.New("bus record not found") + func IsDeliveryType(deliveryType string) bool + type BatchBuilder interface + AllowFailures func() BatchBuilder + Catch func(fn func(ctx context.Context, st BatchState, err error) error) BatchBuilder + Dispatch func(ctx context.Context) (string, error) + Finally func(fn func(ctx context.Context, st BatchState) error) BatchBuilder + Name func(name string) BatchBuilder + OnQueue func(queue string) BatchBuilder + Progress func(fn func(ctx context.Context, st BatchState) error) BatchBuilder + Then func(fn func(ctx context.Context, st BatchState) error) BatchBuilder + type BatchJob struct + Job StoredJob + JobID string + type BatchJobOutcome string + const BatchJobFailed + const BatchJobSucceeded + type BatchRecord struct + AllowFailed bool + BatchID string + CreatedAt time.Time + DispatchID string + Jobs []BatchJob + Name string + Queue string + func (r BatchRecord) Continuations() []ContinuationDefinition + type BatchState struct + AllowFailed bool + BatchID string + Cancelled bool + Completed bool + CreatedAt time.Time + DispatchID string + Failed int + Name string + Pending int + Processed int + Queue string + Total int + UpdatedAt time.Time + type ChainBuilder interface + Catch func(fn func(ctx context.Context, st ChainState, err error) error) ChainBuilder + Dispatch func(ctx context.Context) (string, error) + Finally func(fn func(ctx context.Context, st ChainState) error) ChainBuilder + OnQueue func(queue string) ChainBuilder + type ChainNode struct + Job StoredJob + NodeID string + type ChainRecord struct + ChainID string + CreatedAt time.Time + DispatchID string + Nodes []ChainNode + Queue string + func (r ChainRecord) Continuations() []ContinuationDefinition + type ChainState struct + ChainID string + Completed bool + CreatedAt time.Time + DispatchID string + Failed bool + Failure string + NextIndex int + Nodes []ChainNode + Queue string + UpdatedAt time.Time + type Context struct + Attempt int + BatchID string + ChainID string + DispatchID string + JobID string + JobType string + SchemaVersion int + func NewContext(schemaVersion int, dispatchID, jobID, chainID, batchID string, attempt int, ...) Context + func (c Context) Bind(dst any) error + func (c Context) Continuation() (busruntime.Continuation, bool) + func (c Context) PayloadBytes() []byte + func (c Context) WithContinuation(value busruntime.Continuation) Context + type ContinuationBinder interface + BindContinuations func([]ContinuationDefinition) (any, error) + type ContinuationClaim = continuationClaim + type ContinuationDefinition struct + Job StoredJob + Phase string + type ContinuationReader interface + FindContinuations func(context.Context, string) ([]busruntime.ContinuationOutcome, error) + type ContinuationStore interface + BeginContinuation func(context.Context, continuationClaim, string, time.Time) (continuationClaim, bool, error) + ClaimContinuation func(context.Context, string, string, time.Time, time.Duration) (continuationClaim, bool, error) + ContinuationOutcomes func(context.Context, string) ([]busruntime.ContinuationOutcome, error) + PrepareContinuations func(context.Context) error + RenewContinuation func(context.Context, continuationClaim, time.Time) error + SettleContinuation func(context.Context, continuationClaim, string, time.Time) error + ValidateContinuation func(context.Context, continuationClaim) (bool, error) + type CoordinatorLifecycle interface + CoordinatorDone func() <-chan struct{} + CoordinatorError func() error + StartCoordinator func(context.Context) error + StopCoordinator func(context.Context) error + type DispatchResult struct + DispatchID string + type Engine interface + Batch func(jobs ...Job) BatchBuilder + Chain func(jobs ...Job) ChainBuilder + FindBatch func(ctx context.Context, batchID string) (BatchState, error) + FindChain func(ctx context.Context, chainID string) (ChainState, error) + Prune func(ctx context.Context, before time.Time) error + func NewCoordinator(jobs *Jobs, opts ...Option) Engine + type Event struct + Attempt int + BatchID string + ChainID string + DispatchID string + Duration time.Duration + Err error + EventID string + JobID string + JobKey string + JobType string + Kind EventKind + Queue string + SchemaVersion int + Time time.Time + type EventKind string + const EventBatchCancelled + const EventBatchCompleted + const EventBatchFailed + const EventBatchProgressed + const EventBatchStarted + const EventCallbackFailed + const EventCallbackStarted + const EventCallbackSucceeded + const EventChainAdvanced + const EventChainCompleted + const EventChainFailed + const EventChainStarted + const EventDispatchFailed + const EventDispatchStarted + const EventDispatchSucceeded + const EventJobFailed + const EventJobStarted + const EventJobSucceeded + type Handler func(ctx context.Context, j Context) error + type Job struct + Options JobOptions + Payload any + Type string + func NewJob(jobType string, payload any) Job + func (j Job) Backoff(backoff time.Duration) Job + func (j Job) Delay(delay time.Duration) Job + func (j Job) OnQueue(name string) Job + func (j Job) Retry(max int) Job + func (j Job) Timeout(timeout time.Duration) Job + func (j Job) UniqueFor(ttl time.Duration) Job + type JobConfig struct + Clock func() time.Time + Middlewares []Middleware + Observer Observer + PanicError func(any) error + TranslateError func(error) error + type JobOptions struct + Backoff time.Duration + Delay time.Duration + Queue string + Retry int + Timeout time.Duration + UniqueFor time.Duration + type Jobs struct + func NewJobs(q any, cfg JobConfig) (*Jobs, error) + func (r *Jobs) Dispatch(ctx context.Context, job Job) (DispatchResult, error) + func (r *Jobs) DispatchDirect(ctx context.Context, job StoredJob) (DispatchResult, error) + func (r *Jobs) ExecuteDirect(ctx context.Context, job StoredJob, metadata busruntime.DeliveryMetadata, ...) error + func (r *Jobs) Register(jobType string, handler Handler) + func (r *Jobs) Shutdown(ctx context.Context) error + func (r *Jobs) StartWorkers(ctx context.Context) error + type Middleware interface + Handle func(ctx context.Context, jc Context, next Next) error + type Next func(ctx context.Context, jc Context) error + type Observer interface + Observe func(ctx context.Context, event Event) + func MultiObserver(observers ...Observer) Observer + type ObserverFunc func(ctx context.Context, event Event) + func (f ObserverFunc) Observe(ctx context.Context, event Event) + type Option func(*runtime) + func WithClock(clock func() time.Time) Option + func WithContinuationQueue(queue string) Option + func WithContinuationResource(resource string) Option + func WithMiddleware(middlewares ...Middleware) Option + func WithObserver(observer Observer) Option + func WithStore(store Store) Option + func WithoutEphemeralCallbacks() Option + type ResolvedDelivery struct + BatchID string + ChainID string + DispatchID string + JobID string + JobType string + Payload []byte + func ResolveDelivery(deliveryType string, payload []byte) ResolvedDelivery + type Runtime interface + Dispatch func(ctx context.Context, job Job) (DispatchResult, error) + DispatchDirect func(ctx context.Context, job StoredJob) (DispatchResult, error) + Register func(jobType string, handler Handler) + Shutdown func(ctx context.Context) error + StartWorkers func(ctx context.Context) error + func New(q any, opts ...Option) (Runtime, error) + func NewWithStore(q any, store Store, opts ...Option) (Runtime, error) + type SQLStoreConfig struct + AutoMigrate bool + DB *sql.DB + DSN string + DriverName string + type Store interface + AdvanceChain func(ctx context.Context, chainID string, completedNode string) (next *ChainNode, done bool, err error) + CancelBatch func(ctx context.Context, batchID string) error + CreateBatch func(ctx context.Context, rec BatchRecord) error + CreateChain func(ctx context.Context, rec ChainRecord) error + FailChain func(ctx context.Context, chainID string, cause error) error + GetBatch func(ctx context.Context, batchID string) (BatchState, error) + GetChain func(ctx context.Context, chainID string) (ChainState, error) + MarkBatchJobFailed func(ctx context.Context, batchID, jobID string, cause error) (BatchState, bool, error) + MarkBatchJobStarted func(ctx context.Context, batchID, jobID string) error + MarkBatchJobSucceeded func(ctx context.Context, batchID, jobID string) (BatchState, bool, error) + MarkCallbackInvoked func(ctx context.Context, key string) (bool, error) + Prune func(ctx context.Context, before time.Time) error + func NewMemoryStore() Store + func NewSQLStore(cfg SQLStoreConfig) (Store, error) + func NewSQLStoreWithManagedSchema(cfg SQLStoreConfig) (Store, error) + type StoredJob struct + Options JobOptions + Payload []byte + Type string
Go to main page