Versions in this module Expand all Collapse all v1 v1.2.1 Oct 6, 2026 v1.2.0 Oct 6, 2026 Changes in this version + const DefaultMaxDecodedSize + var ErrDecodedTooLarge = errors.New("amqp: decoded body exceeds the size limit") type Encoder + func NewDeflate(maxDecodedSize int64) Encoder + func NewGzip(maxDecodedSize int64) Encoder v1.1.0 Oct 6, 2026 Changes in this version + var ErrUnconfirmed = errors.New("amqp: channel closed before the message was confirmed") type Config + ConnectTimeout time.Duration v1.0.1 Oct 6, 2026 v1.0.0 Oct 6, 2026 Changes in this version + const AccessRefused + const ChannelError + const CommandInvalid + const ConnectionForced + const ContentTooLarge + const Direct + const Fanout + const FrameError + const Headers + const InternalError + const InvalidPath + const NoConsumers + const NoRoute + const NotAllowed + const NotFound + const NotImplemented + const Persistent + const PreconditionFailed + const ReplySuccess + const ResourceError + const ResourceLocked + const SyntaxError + const Topic + const Transient + const UnexpectedFrame + const Version + var ErrAlreadyAcknowledged = errors.New("amqp: delivery already acknowledged or rejected") + var ErrChannelMax = errors.New("amqp: channel max reached") + var ErrClosed = errors.New("amqp: closed") + var ErrConsumerCancelled = errors.New("amqp: consumer cancelled") + var ErrConsumerCancelledByServer = errors.New("amqp: consumer cancelled by the broker") + var ErrNoConfirmMode = errors.New("amqp: channel is not in confirm mode") + var ErrPublishNacked = errors.New("amqp: message nacked by the broker") + var ErrUnroutable = errors.New("amqp: message could not be routed to any queue") + var ErrUnsupportedContentEncoding = errors.New("amqp: unsupported content encoding") + var ErrUnsupportedContentType = errors.New("amqp: unsupported content type") + func IsCode(err error, code uint16) bool + type Channel struct + func (ch *Channel) BasicAck(deliveryTag uint64, multiple bool) error + func (ch *Channel) BasicCancel(ctx context.Context, consumerTag string) error + func (ch *Channel) BasicConsume(ctx context.Context, queue string, opts ConsumeOptions) (*Consumer, error) + func (ch *Channel) BasicGet(ctx context.Context, queue string, noAck bool) (msg *Delivery, ok bool, err error) + func (ch *Channel) BasicNack(deliveryTag uint64, multiple, requeue bool) error + func (ch *Channel) BasicPublish(ctx context.Context, exchange, routingKey string, msg Publishing) error + func (ch *Channel) BasicPublishConfirm(ctx context.Context, exchange, routingKey string, msg Publishing) (*Confirmation, error) + func (ch *Channel) BasicQos(ctx context.Context, prefetchCount uint16, global bool) error + func (ch *Channel) BasicRecover(ctx context.Context, requeue bool) error + func (ch *Channel) BasicReject(deliveryTag uint64, requeue bool) error + func (ch *Channel) Close() error + func (ch *Channel) CloseReason(code uint16, reason string) error + func (ch *Channel) ConfirmSelect(ctx context.Context) error + func (ch *Channel) Connection() *Connection + func (ch *Channel) Done() <-chan struct{} + func (ch *Channel) Err() error + func (ch *Channel) ExchangeBind(ctx context.Context, destination, source, routingKey string, args Table) error + func (ch *Channel) ExchangeDeclare(ctx context.Context, name, kind string, opts ExchangeDeclareOptions) error + func (ch *Channel) ExchangeDelete(ctx context.Context, name string, ifUnused bool) error + func (ch *Channel) ExchangeUnbind(ctx context.Context, destination, source, routingKey string, args Table) error + func (ch *Channel) Flow(ctx context.Context, active bool) error + func (ch *Channel) ID() uint16 + func (ch *Channel) IsClosed() bool + func (ch *Channel) OnReturn(fn func(*Return)) + func (ch *Channel) QueueBind(ctx context.Context, queue, exchange, routingKey string, args Table) error + func (ch *Channel) QueueDeclare(ctx context.Context, name string, opts QueueDeclareOptions) (QueueInfo, error) + func (ch *Channel) QueueDelete(ctx context.Context, queue string, opts QueueDeleteOptions) (uint32, error) + func (ch *Channel) QueuePurge(ctx context.Context, queue string) (uint32, error) + func (ch *Channel) QueueUnbind(ctx context.Context, queue, exchange, routingKey string, args Table) error + func (ch *Channel) TxCommit(ctx context.Context) error + func (ch *Channel) TxRollback(ctx context.Context) error + func (ch *Channel) TxSelect(ctx context.Context) error + func (ch *Channel) WaitForConfirms(ctx context.Context) error + type Client struct + func NewClient(ctx context.Context, url string, opts *ClientOptions) (*Client, error) + func (c *Client) Close() error + func (c *Client) Connection(ctx context.Context) (*Connection, error) + func (c *Client) DefaultExchange() *Exchange + func (c *Client) DirectExchange(ctx context.Context, name string) (*Exchange, error) + func (c *Client) Done() <-chan struct{} + func (c *Client) Err() error + func (c *Client) Exchange(ctx context.Context, name, kind string, opts *ExchangeOptions) (*Exchange, error) + func (c *Client) FanoutExchange(ctx context.Context, name string) (*Exchange, error) + func (c *Client) HeadersExchange(ctx context.Context, name string) (*Exchange, error) + func (c *Client) Publish(ctx context.Context, exchange, routingKey string, msg Message) error + func (c *Client) Queue(ctx context.Context, name string, opts *QueueOptions) (*Queue, error) + func (c *Client) RPCCall(ctx context.Context, queue string, msg Message) (*Delivery, error) + func (c *Client) RPCClient(ctx context.Context) (*RPCClient, error) + func (c *Client) RPCServer(ctx context.Context, queue string, handler RPCHandler, opts *SubscribeOptions) (*Subscription, error) + func (c *Client) TopicExchange(ctx context.Context, name string) (*Exchange, error) + type ClientOptions struct + DefaultContentEncoding string + DefaultContentType string + MaxReconnectInterval time.Duration + MaxRetries int + OnConnect func(ctx context.Context, c *Client) error + OnDisconnect func(err error) + OnFailed func(err error) + OnReturn func(*Return) + ReconnectInterval time.Duration + type Codecs struct + func DefaultCodecs() *Codecs + func NewCodecs() *Codecs + func (c *Codecs) Marshal(v any, p *Properties) ([]byte, error) + func (c *Codecs) RegisterEncoder(contentEncoding string, e Encoder) *Codecs + func (c *Codecs) RegisterSerializer(contentType string, s Serializer) *Codecs + func (c *Codecs) Unmarshal(body []byte, p *Properties, v any) error + type Config struct + ChannelMax uint16 + ClientProperties Table + Codecs *Codecs + ConnectionName string + Dial func(ctx context.Context, network, addr string) (net.Conn, error) + FrameMax uint32 + Heartbeat time.Duration + Logger *slog.Logger + OnBlocked func(reason string) + OnUnblocked func() + TLSConfig *tls.Config + type Confirmation struct + func (c *Confirmation) Acked() bool + func (c *Confirmation) DeliveryTag() uint64 + func (c *Confirmation) Done() <-chan struct{} + func (c *Confirmation) Wait(ctx context.Context) error + type Connection struct + func Dial(ctx context.Context, uri string, cfg *Config) (*Connection, error) + func DialURI(ctx context.Context, u URI, cfg *Config) (*Connection, error) + func (c *Connection) Blocked() bool + func (c *Connection) Channel(ctx context.Context) (*Channel, error) + func (c *Connection) ChannelMax() uint16 + func (c *Connection) Close() error + func (c *Connection) CloseReason(reason string) error + func (c *Connection) Done() <-chan struct{} + func (c *Connection) Err() error + func (c *Connection) FrameMax() uint32 + func (c *Connection) Heartbeat() time.Duration + func (c *Connection) IsClosed() bool + func (c *Connection) LocalAddr() net.Addr + func (c *Connection) RemoteAddr() net.Addr + func (c *Connection) ServerProperties() Table + func (c *Connection) UpdateSecret(ctx context.Context, secret, reason string) error + type ConsumeOptions struct + Args Table + Exclusive bool + NoAck bool + NoLocal bool + Tag string + type Consumer struct + func (c *Consumer) Cancel(ctx context.Context) error + func (c *Consumer) Channel() *Channel + func (c *Consumer) Deliveries() <-chan *Delivery + func (c *Consumer) Done() <-chan struct{} + func (c *Consumer) Err() error + func (c *Consumer) Next(ctx context.Context) (*Delivery, error) + func (c *Consumer) Queue() string + func (c *Consumer) Tag() string + type Decimal struct + Scale uint8 + Value int32 + func (d Decimal) String() string + type Delivery struct + Body []byte + ConsumerTag string + DeliveryTag uint64 + Exchange string + MessageCount uint32 + Redelivered bool + RoutingKey string + func (d *Delivery) Ack() error + func (d *Delivery) Acknowledged() bool + func (d *Delivery) Channel() *Channel + func (d *Delivery) Decode(v any) error + func (d *Delivery) Nack(requeue bool) error + func (d *Delivery) Reject(requeue bool) error + type Encoder interface + Decode func(data []byte) ([]byte, error) + Encode func(data []byte) ([]byte, error) + var Deflate Encoder = &deflateEncoder{} + var Gzip Encoder = &gzipEncoder{} + type Error struct + ClassID uint16 + Code uint16 + Connection bool + MethodID uint16 + Reason string + Server bool + func (e *Error) Error() string + func (e *Error) Is(target error) bool + type Exchange struct + func (x *Exchange) Bind(ctx context.Context, source, routingKey string, args Table) error + func (x *Exchange) Delete(ctx context.Context, ifUnused bool) error + func (x *Exchange) Kind() string + func (x *Exchange) Name() string + func (x *Exchange) Publish(ctx context.Context, routingKey string, msg Message) error + func (x *Exchange) Unbind(ctx context.Context, source, routingKey string, args Table) error + type ExchangeDeclareOptions struct + Args Table + AutoDelete bool + Durable bool + Internal bool + Passive bool + type ExchangeOptions struct + Args Table + AutoDelete bool + Durable bool + Internal bool + Passive bool + type Handler func(ctx context.Context, d *Delivery) error + type Message struct + Body any + Mandatory bool + type Properties struct + AppID string + ContentEncoding string + ContentType string + CorrelationID string + DeliveryMode uint8 + Expiration string + Headers Table + MessageID string + Priority uint8 + ReplyTo string + Timestamp time.Time + Type string + UserID string + type Publishing struct + Body []byte + Mandatory bool + type Queue struct + func (q *Queue) Bind(ctx context.Context, exchange, bindingKey string, args Table) error + func (q *Queue) Delete(ctx context.Context, opts QueueDeleteOptions) (n uint32, err error) + func (q *Queue) Get(ctx context.Context, noAck bool) (msg *Delivery, ok bool, err error) + func (q *Queue) Name() string + func (q *Queue) Publish(ctx context.Context, msg Message) error + func (q *Queue) Purge(ctx context.Context) (n uint32, err error) + func (q *Queue) Subscribe(ctx context.Context, handler Handler, opts *SubscribeOptions) (*Subscription, error) + func (q *Queue) Unbind(ctx context.Context, exchange, bindingKey string, args Table) error + type QueueDeclareOptions struct + Args Table + AutoDelete bool + Durable bool + Exclusive bool + Passive bool + type QueueDeleteOptions struct + IfEmpty bool + IfUnused bool + type QueueInfo struct + ConsumerCount uint32 + MessageCount uint32 + Name string + type QueueOptions struct + Args Table + AutoDelete bool + Durable bool + Exclusive bool + Passive bool + type RPCClient struct + func (r *RPCClient) Call(ctx context.Context, queue string, msg Message) (*Delivery, error) + func (r *RPCClient) Close() error + type RPCHandler func(ctx context.Context, req *Delivery) (Message, error) + type Return struct + Body []byte + Exchange string + ReplyCode uint16 + ReplyText string + RoutingKey string + type Serializer interface + Marshal func(v any) ([]byte, error) + Unmarshal func(data []byte, v any) error + var JSON Serializer = jsonSerializer{} + var Text Serializer = textSerializer{} + type SubscribeOptions struct + Args Table + ConsumerTag string + Exclusive bool + NoAck bool + Prefetch uint16 + Workers int + type Subscription struct + func (s *Subscription) Cancel(ctx context.Context) error + func (s *Subscription) ConsumerTag() string + func (s *Subscription) Done() <-chan struct{} + func (s *Subscription) Err() error + func (s *Subscription) Queue() *Queue + type Table map[string]any + type URI struct + AuthMechanism string + ChannelMax uint16 + ConnectTimeout time.Duration + ConnectionName string + FrameMax uint32 + Heartbeat time.Duration + Host string + InsecureTLS bool + Password string + Port int + ServerName string + TLS bool + Username string + Vhost string + func ParseURI(s string) (URI, error) + func (u URI) Addr() string + func (u URI) String() string
Go to main page