Versions in this module Expand all Collapse all v0 v0.1.0 Jul 14, 2026 Changes in this version + const DefaultName + const DefaultRetryFactor + const DefaultRetryInitialDelay + const DefaultRetryMaxDelay + const HeaderAttempts + const HeaderError + const HeaderOriginTopic + var ErrNoTopic = errors.New("smsg: no topic resolved for message") + var ErrNotStarted = errors.New("smsg: bus not started") + var ErrStopped = errors.New("smsg: bus stopped") + func DefaultTopicNamer(t reflect.Type) string + type Builder struct + func New() *Builder + func (b *Builder) AddConsumer(c ConsumerRegistration, opts ...ConsumerOption) *Builder + func (b *Builder) Build() (*Bus, error) + func (b *Builder) MustBuild() *Bus + func (b *Builder) WithLogger(l Logger) *Builder + func (b *Builder) WithName(name string) *Builder + func (b *Builder) WithObserver(o Observer) *Builder + func (b *Builder) WithRetry(p RetryPolicy) *Builder + func (b *Builder) WithSerializer(s Serializer) *Builder + func (b *Builder) WithTopicNamer(fn TopicNamer) *Builder + func (b *Builder) WithTransport(t Transport) *Builder + type Bus struct + func (b *Bus) Name() string + func (b *Bus) Publish(ctx context.Context, msg any, opts ...PublishOption) error + func (b *Bus) Ready() <-chan struct{} + func (b *Bus) Start(ctx context.Context) error + func (b *Bus) Stop(ctx context.Context) error + type Consumer interface + Consume func(ctx context.Context, m Message[T]) error + type ConsumerFunc func(ctx context.Context, m Message[T]) error + func (f ConsumerFunc[T]) Consume(ctx context.Context, m Message[T]) error + type ConsumerOption func(*consumerConfig) + func Concurrency(n int) ConsumerOption + func ConsumerMeta(key, value string) ConsumerOption + func DeadLetter(topic string) ConsumerOption + func Group(name string) ConsumerOption + func OnExhausted(a ExhaustedAction) ConsumerOption + func Retry(p RetryPolicy) ConsumerOption + func Topic(name string) ConsumerOption + type ConsumerRegistration struct + func For[T any](c Consumer[T]) ConsumerRegistration + func Handle[T any](fn func(ctx context.Context, m Message[T]) error) ConsumerRegistration + type Delivery struct + Attempt int + Envelope *Envelope + type DeliveryHandler func(ctx context.Context, d *Delivery) error + type Envelope struct + Body []byte + ContentType string + CorrelationID string + Headers map[string]string + MessageID string + MessageType string + Metadata map[string]string + Timestamp time.Time + Topic string + func (e *Envelope) Clone() *Envelope + type ExhaustedAction int + const ExhaustedDeadLetter + const ExhaustedDrop + const ExhaustedRequeue + type Logger interface + Debug func(template string, args ...any) + Error func(err error, template string, args ...any) + Information func(template string, args ...any) + Warning func(template string, args ...any) + func SlogLogger(l *slog.Logger) Logger + type Message struct + Attempt int + Body T + CorrelationID string + Headers map[string]string + ID string + Timestamp time.Time + Topic string + type Observer struct + BusStarted func() + BusStopped func(err error) + Consumed func(topic, group, messageID string, attempt int, elapsed time.Duration, err error) + DeadLettered func(topic, group, dlqTopic, messageID string, err error) + Dropped func(topic, group, messageID string, err error) + Published func(topic string, env *Envelope, elapsed time.Duration, err error) + RetryScheduled func(topic, group, messageID string, attempt int, delay time.Duration, err error) + SubscriptionStarted func(topic, group string) + SubscriptionStopped func(topic, group string, err error) + type PublishOption func(*publishConfig) + func Meta(key, value string) PublishOption + func ToTopic(topic string) PublishOption + func WithCorrelationID(id string) PublishOption + func WithHeader(key, value string) PublishOption + func WithMessageID(id string) PublishOption + type RetryPolicy struct + Factor float64 + InitialDelay time.Duration + MaxAttempts int + MaxDelay time.Duration + type Serializer interface + ContentType func() string + Deserialize func(data []byte, v any) error + Serialize func(v any) ([]byte, error) + func JSON() Serializer + type Subscription interface + Close func(ctx context.Context) error + Done func() <-chan struct{} + Err func() error + type SubscriptionSpec struct + Concurrency int + Group string + Metadata map[string]string + Topic string + type TopicNamer func(t reflect.Type) string + type Transport interface + Close func(ctx context.Context) error + Connect func(ctx context.Context) error + Name func() string + Publish func(ctx context.Context, topic string, env *Envelope) error + Subscribe func(ctx context.Context, spec SubscriptionSpec, h DeliveryHandler) (Subscription, error)
Go to main page