messaging

Apr 25, 2026

Changes in this version

+ func EncodeMessagePayload(msg *Message) ([]byte, error)

type Message

+ func DecodeMessagePayload(payload []byte) (*Message, bool, error)

+ func (m *Message) IsSettled() bool

Dec 3, 2025

Changes in this version

+ func ParseJSON(msg *Message, v interface{}) error

+ func PublishEvent(ctx context.Context, publisher Publisher, event string, body []byte) error

+ func PublishJSON(ctx context.Context, publisher Publisher, topic string, data interface{}) error

+ func PublishTask(ctx context.Context, publisher Publisher, taskType string, body []byte) error

+ func RegisterProvider(provider Provider, factory EventBusFactory)

+ func SubscribeEvent(subscriber Subscriber, event string, serviceName string, handler Handler) error

+ func SubscribeTask(subscriber Subscriber, taskType string, workerGroup string, handler Handler) error

+ type AuditLogger interface

+ Log func(event string, msg *Message, err error)

+ type BatchProcessor interface

+ Add func(msg *Message) error

+ Flush func() error

+ type CircuitBreaker interface

+ Call func(fn func() error) error

+ State func() string

+ type Config struct

+ NSQ NSQConfig

+ Provider Provider

+ RabbitMQ RabbitMQConfig

+ func DefaultConfig() *Config

+ type DeduplicationStore interface

+ Exists func(uuid string) bool

+ Mark func(uuid string, ttl time.Duration) error

+ type EventBus interface

+ Close func() error

+ Health func() error

+ Publisher func() Publisher

+ Router func() *Router

+ Subscriber func() Subscriber

+ func MustNewEventBus(config *Config) EventBus

+ func NewEventBus(config *Config) (EventBus, error)

+ type EventBusFactory func(config *Config) (EventBus, error)

+ type Handler func(ctx context.Context, msg *Message) error

+ func NewJSONHandler(fn func(ctx context.Context, data interface{}) error, dataType interface{}) Handler

+ func NewSimpleHandler(fn func([]byte) error) Handler

+ func WrapHandler(oldHandler func(topic string, data []byte) error) Handler

+ type Message struct

+ Attempts uint16

+ Channel string

+ Metadata map[string]string

+ Payload []byte

+ Timestamp int64

+ Topic string

+ UUID string

+ func NewMessage(uuid string, payload []byte) *Message

+ func (m *Message) Ack() error

+ func (m *Message) Body() []byte

+ func (m *Message) ID() string

+ func (m *Message) Nack() error

+ func (m *Message) SetAckFunc(ack func() error)

+ func (m *Message) SetNackFunc(nack func() error)

+ type MetricsCollector interface

+ IncrementFailed func(topic string)

+ IncrementProcessed func(topic string)

+ RecordDuration func(topic string, duration time.Duration)

+ type Middleware func(Handler) Handler

+ func AuditMiddleware(logger AuditLogger) Middleware

+ func BatchMiddleware(batchSize int, flushInterval time.Duration, ...) Middleware

+ func CircuitBreakerMiddleware(breaker CircuitBreaker) Middleware

+ func DeduplicationMiddleware(store DeduplicationStore, ttl time.Duration) Middleware

+ func FilterMiddleware(predicate func(*Message) bool) Middleware

+ func LoggerMiddleware(logger *log.Logger) Middleware

+ func MetricsMiddleware(collector MetricsCollector) Middleware

+ func PriorityMiddleware(getPriority func(*Message) int) Middleware

+ func RateLimitMiddleware(limiter RateLimiter, mode string) Middleware

+ func RecoverMiddleware(logger *log.Logger) Middleware

+ func RetryMiddleware(maxRetries int, delay time.Duration) Middleware

+ func TimeoutMiddleware(timeout time.Duration) Middleware

+ func TracingMiddleware() Middleware

+ type NSQConfig struct

+ DialTimeout time.Duration

+ LookupdAddrs []string

+ MaxAttempts uint16

+ MaxInFlight int

+ MsgTimeout time.Duration

+ NSQdAddr string

+ ReadTimeout time.Duration

+ RequeueDelay time.Duration

+ WriteTimeout time.Duration

+ func DefaultNSQConfig() NSQConfig

+ type Provider string

+ const ProviderNSQ

+ const ProviderRabbitMQ

+ func GetRegisteredProviders() []Provider

+ type Publisher interface

+ Close func() error

+ Publish func(ctx context.Context, topic string, body []byte) error

+ PublishMessage func(ctx context.Context, topic string, msg *Message) error

+ type RabbitMQConfig struct

+ AutoDelete bool

+ AutoReconnect bool

+ ConnectionTimeout time.Duration

+ Durable bool

+ ExchangeType string

+ Exclusive bool

+ HeartbeatInterval time.Duration

+ Host string

+ MaxChannels int

+ MaxReconnectAttempts int

+ Password string

+ PersistentMessages bool

+ Port int

+ PrefetchCount int

+ PrefetchSize int

+ ReconnectDelay time.Duration

+ URL string

+ Username string

+ VHost string

+ func DefaultRabbitMQConfig() RabbitMQConfig

+ func (c *RabbitMQConfig) BuildURL() string

+ type RateLimiter interface

+ Allow func() bool

+ Wait func(ctx context.Context) error

+ type Router struct

+ func NewRouter(subscriber Subscriber) *Router

+ func (r *Router) AddHandler(topic, channel string, handler Handler)

+ func (r *Router) AddHandlerWithMiddleware(topic, channel string, handler Handler, middlewares ...Middleware)

+ func (r *Router) AddMiddleware(mw Middleware)

+ func (r *Router) Run(ctx context.Context) error

+ func (r *Router) Stop()

+ type SimpleCircuitBreaker struct

+ func NewSimpleCircuitBreaker(maxFailures int, timeout time.Duration) *SimpleCircuitBreaker

+ func (cb *SimpleCircuitBreaker) Call(fn func() error) error

+ func (cb *SimpleCircuitBreaker) State() string

+ type Subscriber interface

+ Close func() error

+ Stop func()

+ Subscribe func(topic, channel string, handler Handler) error

+ SubscribeWithMiddleware func(topic, channel string, handler Handler, middlewares ...Middleware) error

+ type TaskPublisher struct

+ func NewTaskPublisher(publisher Publisher) *TaskPublisher

+ func (tp *TaskPublisher) Close() error

+ func (tp *TaskPublisher) Publish(ctx context.Context, taskType string, body []byte) error

+ type TaskSubscriber struct

+ func NewTaskSubscriber(subscriber Subscriber, workerGroup string) *TaskSubscriber

+ func (ts *TaskSubscriber) Close() error

+ func (ts *TaskSubscriber) Stop()

+ func (ts *TaskSubscriber) Subscribe(taskType string, handler Handler) error

+ type TokenBucketLimiter struct

+ func NewTokenBucketLimiter(capacity int, rate time.Duration) *TokenBucketLimiter

+ func (l *TokenBucketLimiter) Allow() bool

+ func (l *TokenBucketLimiter) Wait(ctx context.Context) error