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

messaging

package module

Go to main page

Versions in this module

v0

v0.3.0

Nov 11, 2026

Changes in this version

+ type Setup struct

+ Broker Broker

+ Config Config

+ Gate Gate

+ Inbox inbox.Engine

+ Logger *slog.Logger

+ OnFailure Hook

+ Outbox outbox.Engine

+ Shutdown time.Duration

+ func (s Setup) Build() (*Runtime, error)

v0.2.0

Oct 14, 2026

Changes in this version

+ type Gate interface

+ Shared func(ctx context.Context) (release func(), err error)

type Option

+ func Gated(g Gate) Option

v0.1.0

Oct 13, 2026

Changes in this version

+ const DefaultAckWait

+ const DefaultDuplicates

+ const DefaultRelayPoll

+ const DefaultRelayTimeout

+ func Attempt(ctx context.Context) int

+ func CheckName(s string) error

+ func CheckType(t string) error

+ func Encode(e event.Event) (event.Header, []byte, error)

+ func WithAttempt(ctx context.Context, attempt int) context.Context

+ type Broker interface

+ Subscribe func(sub Subscription, report Report) (reactor.Source[event.Event], error)

+ type Claim = func(ctx context.Context, tx *sqlate.Tx) (first bool, err error)

+ type Config struct

+ RelayPoll libconfig.Duration

+ RelayTimeout libconfig.Duration

+ Source string

+ func (c *Config) Finalize(envPrefix string) error

+ func (c *Config) Merge(src *Config)

+ func (c Config) Validate() error

+ type Failure struct

+ Attempt int

+ Consumer string

+ Err error

+ EventID string

+ EventSource string

+ EventType string

+ Kind FailureKind

+ type FailureKind string

+ const FailureExhausted

+ const FailureQuarantine

+ const FailureRelay

+ const FailureSource

+ type Hook func(ctx context.Context, f Failure)

+ type Option func(*Runtime)

+ func OnFailure(h Hook) Option

+ type Publisher interface

+ Publish func(ctx context.Context, e event.Event) error

+ type Report func(ctx context.Context, err error)

+ type Runtime struct

+ Recorder *event.Recorder[*sqlate.Tx]

+ func New(cfg Config, broker Broker, outboxEngine outbox.Engine, ...) (*Runtime, error)

+ func (r *Runtime) Consume[T any](sub Subscription, fn func(ctx context.Context, data T, claim Claim) error) (*reactor.Reactor[event.Event], error)

+ func (r *Runtime) Relay(db sqlate.Beginner) *reactor.Reactor[event.Event]

+ type Start int

+ const StartAll

+ const StartNew

+ type Subscription struct

+ AckWait time.Duration

+ MaxDeliver int

+ Name string

+ RetryDelay time.Duration

+ Start Start

+ Types []string

+ func (sub Subscription) Matches(t string) bool

+ func (sub Subscription) Normalize() Subscription

+ func (sub Subscription) Validate() error