pubsub

Mar 26, 2026

Changes in this version

+ var ErrConsumerAlreadyExists = errors.New("consumer with this name already exists")

+ var ErrInvalidBind = errors.New("invalid queue binding")

+ var ErrInvalidConsumer = errors.New("invalid consumer configuration")

+ var ErrInvalidDelay = errors.New("invalid delay option value")

+ var ErrInvalidExchange = errors.New("invalid exchange declaration")

+ var ErrInvalidQueue = errors.New("invalid queue declaration")

+ type Logger interface

+ Printf func(format string, v ...any)

type Option

+ func WithLogger(l Logger) Option

+ func WithReInitDelay(d time.Duration) Option

+ func WithReconnectDelay(d time.Duration) Option

+ func WithResendDelay(d time.Duration) Option

type Session

+ func (session *Session) PublishToWithContext(ctx context.Context, exchange, key string, message []byte) error

+ func (session *Session) PublishWithContext(ctx context.Context, message []byte) error

+ func (session *Session) RemoveConsumer(tag string) error

+ func (session *Session) SubscribeToWithTag(queue string, handler func(Delivery) error) (string, error)

+ func (session *Session) SubscribeWithTag(handler func(Delivery) error) (string, error)

+ func (session *Session) UnsafePublishTo(exchange, key string, message []byte) error

+ func (session *Session) UnsafePublishToWithContext(ctx context.Context, exchange, key string, message []byte) error

+ func (session *Session) WaitReady(ctx context.Context) error

Jan 19, 2022

Changes in this version

+ type Delivery struct

+ Body []byte

+ Exchange string

+ RoutingKey string

Jan 16, 2022

Changes in this version

type Bind

+ IsUsageDefault bool

type Exchange

+ IsUsageDefault bool

type Queue

+ IsUsageDefault bool

type Session

+ func (session *Session) ExchangeDeclare(ex Exchange) error

+ func (session *Session) QueueBind(b Bind) error

Jan 8, 2022

Changes in this version

+ var ErrNotSetDefaultExchange = errors.New("default exchange or key is not set")

+ var ErrNotSetDefaultQueue = errors.New("default queue is not set")

type Session

+ func (session *Session) AddConsumer(c *Consumer) error

+ func (session *Session) Publish(message []byte) error

+ func (session *Session) PublishTo(exchange, key string, message []byte) error

+ func (session *Session) SubscribeTo(queue string, handler func([]byte) error) error

+ func (session *Session) UnsafePublish(message []byte, exchange, key string) error

Jan 7, 2022

Changes in this version

+ var ErrAlreadyClosed = errors.New("already closed: not connected to the server")

+ var ErrNotConnected = errors.New("not connected to a server")

+ var ErrShutdown = errors.New("session is shutting down")

+ type Bind struct

+ ExchangeName string

+ Key string

+ QueueName string

+ type Consumer struct

+ AutoAck bool

+ Exclusive bool

+ Handler EventHandler

+ Name string

+ QueueName string

+ type EventHandler func([]byte) error

+ type Exchange struct

+ AutoDelete bool

+ Durable bool

+ Internal bool

+ Kind string

+ Name string

+ type Option func(*Session)

+ func WithDeclare(ex Exchange, q Queue, b Bind) Option

+ type Queue struct

+ AutoDelete bool

+ Durable bool

+ Exclusive bool

+ Name string

+ type Session struct

+ func New(addr string, opts ...Option) *Session

+ func (session *Session) ChannelClose() error

+ func (session *Session) Close() error

+ func (session *Session) Push(data []byte) error

+ func (session *Session) QueueDeclare(q Queue) *Session

+ func (session *Session) Stream(c *Consumer) (<-chan amqp.Delivery, error)

+ func (session *Session) Subscribe(handler func([]byte) error) error

+ func (session *Session) UnsafePush(data []byte) error