Why Go
- Case Studies
  
  Common problems companies solve with Go
- Use Cases
  
  Stories about how and why companies use Go
- Security
  
  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

bus

package

Go to main page

Versions in this module

v0

v0.1.7

Apr 6, 2026

v0.1.6

Mar 16, 2026

Changes in this version

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

+ var ErrOverlapping = errors.New("bus job overlap prevented")

+ var ErrRateLimited = errors.New("bus job rate limited")

+ var ErrSkipped = errors.New("bus job skipped by middleware")

+ 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 wireJob

+ JobID string

+ type BatchRecord struct

+ AllowFailed bool

+ BatchID string

+ CreatedAt time.Time

+ DispatchID string

+ Jobs []BatchJob

+ Name string

+ Queue string

+ type BatchSpec struct

+ JobTypes []string

+ 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 Bus interface

+ Batch func(jobs ...Job) BatchBuilder

+ Chain func(jobs ...Job) ChainBuilder

+ Dispatch func(ctx context.Context, job Job) (DispatchResult, error)

+ 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

+ Register func(jobType string, handler Handler)

+ Shutdown func(ctx context.Context) error

+ StartWorkers func(ctx context.Context) error

+ func New(q any, opts ...Option) (Bus, error)

+ func NewWithStore(q any, store Store, opts ...Option) (Bus, error)

+ 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 wireJob

+ NodeID string

+ type ChainRecord struct

+ ChainID string

+ CreatedAt time.Time

+ DispatchID string

+ Nodes []ChainNode

+ Queue string

+ 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 (c Context) Bind(dst any) error

+ func (c Context) PayloadBytes() []byte

+ type DispatchResult struct

+ DispatchID string

+ type Event struct

+ Attempt int

+ BatchID string

+ ChainID string

+ DispatchID string

+ Duration time.Duration

+ Err error

+ EventID string

+ JobID 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 FailOnError struct

+ When func(err error) bool

+ func (f FailOnError) Handle(ctx context.Context, jc Context, next Next) error

+ type Fake struct

+ func NewFake() *Fake

+ func (f *Fake) AssertBatchCount(t testing.TB, n int)

+ func (f *Fake) AssertBatched(t testing.TB, predicate func(spec BatchSpec) bool)

+ func (f *Fake) AssertChained(t testing.TB, expected []string)

+ func (f *Fake) AssertCount(t testing.TB, n int)

+ func (f *Fake) AssertDispatched(t testing.TB, jobType string)

+ func (f *Fake) AssertDispatchedOn(t testing.TB, queueName, jobType string)

+ func (f *Fake) AssertDispatchedTimes(t testing.TB, jobType string, n int)

+ func (f *Fake) AssertNotDispatched(t testing.TB, jobType string)

+ func (f *Fake) AssertNothingBatched(t testing.TB)

+ func (f *Fake) AssertNothingDispatched(t testing.TB)

+ func (f *Fake) Batch(jobs ...Job) BatchBuilder

+ func (f *Fake) Chain(jobs ...Job) ChainBuilder

+ func (f *Fake) Dispatch(_ context.Context, job Job) (DispatchResult, error)

+ func (f *Fake) FindBatch(context.Context, string) (BatchState, error)

+ func (f *Fake) FindChain(context.Context, string) (ChainState, error)

+ func (f *Fake) Prune(context.Context, time.Time) error

+ func (f *Fake) Register(string, Handler)

+ func (f *Fake) Shutdown(context.Context) error

+ func (f *Fake) StartWorkers(context.Context) error

+ 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 JobOptions struct

+ Backoff time.Duration

+ Delay time.Duration

+ Queue string

+ Retry int

+ Timeout time.Duration

+ UniqueFor time.Duration

+ type Lock interface

+ Release func(ctx context.Context) error

+ type Locker interface

+ Acquire func(ctx context.Context, key string, ttl time.Duration) (Lock, bool, error)

+ type Middleware interface

+ Handle func(ctx context.Context, jc Context, next Next) error

+ type MiddlewareFunc func(ctx context.Context, jc Context, next Next) error

+ func (f MiddlewareFunc) Handle(ctx context.Context, jc Context, next Next) error

+ type Next func(ctx context.Context, jc Context) error

+ type Observer interface

+ Observe func(event Event)

+ func MultiObserver(observers ...Observer) Observer

+ type ObserverFunc func(event Event)

+ func (f ObserverFunc) Observe(event Event)

+ type Option func(*runtime)

+ func WithClock(clock func() time.Time) Option

+ func WithMiddleware(middlewares ...Middleware) Option

+ func WithObserver(observer Observer) Option

+ func WithStore(store Store) Option

+ type RateLimit struct

+ Key func(ctx context.Context, jc Context) string

+ Limiter RateLimiter

+ func (r RateLimit) Handle(ctx context.Context, jc Context, next Next) error

+ type RateLimiter interface

+ Allow func(ctx context.Context, key string) (allowed bool, retryAfter time.Duration, err error)

+ type RetryPolicy struct

+ func (RetryPolicy) Handle(ctx context.Context, jc Context, next Next) error

+ type SQLStoreConfig struct

+ AutoMigrate bool

+ DB *sql.DB

+ DSN string

+ DriverName string

+ type SkipWhen struct

+ Predicate func(ctx context.Context, jc Context) bool

+ func (s SkipWhen) Handle(ctx context.Context, jc Context, next Next) error

+ 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)

+ type WithoutOverlapping struct

+ Key func(ctx context.Context, jc Context) string

+ Locker Locker

+ TTL time.Duration

+ func (w WithoutOverlapping) Handle(ctx context.Context, jc Context, next Next) error

v0.1.5

Mar 3, 2026

v0.1.4

Mar 1, 2026

v0.1.3

Feb 27, 2026

v0.1.2

Feb 27, 2026

v0.1.1

Feb 27, 2026

v0.1.0

Feb 25, 2026