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
- API
  
  Reference documentation for the pkg.go.dev API
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

dispatch

package

Go to main page

Versions in this module

v1

v1.7.1

Aug 19, 2026

v1.7.0

Aug 19, 2026

v1.6.0

Aug 19, 2026

v1.5.0

Aug 19, 2026

Changes in this version

type Housekeeper

+ DefaultPartitionRows func(ctx context.Context) (int64, error)

+ DropExpiredPartitions func(ctx context.Context, retention time.Duration) ([]string, error)

+ EnsurePartitions func(ctx context.Context, ahead int) ([]string, error)

+ IsPartitioned func(ctx context.Context) (bool, error)

type JanitorEmitter

+ Partitions func(ctx context.Context, ev events.Partitions)

+ type Option func(*Pipeline)

+ func WithTracer(t *tracing.Tracer) Option

v1.4.0

Aug 19, 2026

v1.3.0

Aug 19, 2026

Changes in this version

type Emitter

+ Breaker func(ctx context.Context, ev events.Breaker)

+ type Result struct

+ Claimed int

+ Deferred int

+ Delivered int

v1.2.0

Aug 19, 2026

v1.1.0

Aug 19, 2026

v1.0.0

Aug 19, 2026

Changes in this version

+ type DeadLetter struct

+ func NewDeadLetter(fetch Fetcher, router Router, emitter DeadLetterEmitter, cfg config.Config, ...) *DeadLetter

+ func (d *DeadLetter) Handle(ctx context.Context, ev events.Iteration) error

+ type DeadLetterEmitter interface

+ DeadLetter func(ctx context.Context, ev events.DeadLetter)

+ type Emitter interface

+ Iteration func(ctx context.Context, ev events.Iteration)

+ type Fetcher interface

+ FetchByIDs func(ctx context.Context, ids []string) ([]core.Message, error)

+ type Housekeeper interface

+ Purge func(ctx context.Context, retention time.Duration, limit int) (int64, error)

+ Reclaim func(ctx context.Context, limit int) ([]store.Reclaimed, error)

+ Stats func(ctx context.Context) (store.Stats, error)

+ TryLock func(ctx context.Context, class, key int32) (release func(), ok bool, err error)

+ type Janitor struct

+ func NewJanitor(st Housekeeper, emitter JanitorEmitter, cfg config.Config, log *slog.Logger) *Janitor

+ func (j *Janitor) ReclaimExpired(ctx context.Context)

+ func (j *Janitor) Run(ctx context.Context) error

+ func (j *Janitor) SampleStats(ctx context.Context)

+ func (j *Janitor) Sweep(ctx context.Context)

+ type JanitorEmitter interface

+ Reclaimed func(ctx context.Context, ev events.Reclaimed)

+ Retention func(ctx context.Context, ev events.Retention)

+ Stats func(ctx context.Context, ev events.Stats)

+ type Listener struct

+ func NewListener(pool *pgxpool.Pool, cfg config.DispatchConfig, pipelines []Waker, ...) *Listener

+ func (l *Listener) Run(ctx context.Context) error

+ type Pipeline struct

+ func New(stream string, st Store, router Router, emitter Emitter, cfg config.Config, ...) *Pipeline

+ func (p *Pipeline) Run(ctx context.Context) error

+ func (p *Pipeline) RunOnce(ctx context.Context) (int, error)

+ func (p *Pipeline) Stream() string

+ func (p *Pipeline) Wake()

+ type Router interface

+ DriverFor func(stream string) (string, bool)

+ Publish func(ctx context.Context, stream string, msgs []core.Message) []error

+ type Store interface

+ Ack func(ctx context.Context, ids []string, token string) (store.AckResult, error)

+ Claim func(ctx context.Context, stream string, limit int, lease core.Lease) ([]core.Message, error)

+ Nack func(ctx context.Context, outcomes []core.Outcome, token string, maxAttempts int) (store.NackResult, error)

+ ReleaseLease func(ctx context.Context, ids []string, token string) (int, error)

+ type Waker interface

+ Stream func() string

+ Wake func()