amqp

Documentation

Index

• type Channel
  • func (ch *Channel) Confirm(noWait bool) error
  • func (ch *Channel) Consume(queue, consumer string, opt wabbit.Option) (<-chan wabbit.Delivery, error)
  • func (ch *Channel) ExchangeDeclare(name, kind string, opt wabbit.Option) error
  • func (ch *Channel) ExchangeDeclarePassive(name, kind string, opt wabbit.Option) error
  • func (ch *Channel) NotifyClose(c chan wabbit.Error) chan wabbit.Error
  • func (ch *Channel) NotifyPublish(confirm chan wabbit.Confirmation) chan wabbit.Confirmation
  • func (ch *Channel) Publish(exc, route string, msg []byte, opt wabbit.Option) error
  • func (ch *Channel) Qos(prefetchCount, prefetchSize int, global bool) error
  • func (ch *Channel) QueueBind(name, key, exchange string, opt wabbit.Option) error
  • func (ch *Channel) QueueDeclare(name string, opt wabbit.Option) (wabbit.Queue, error)
  • func (ch *Channel) QueueDeclarePassive(name string, opt wabbit.Option) (wabbit.Queue, error)
  • func (ch *Channel) QueueDelete(name string, opt wabbit.Option) (int, error)
  • func (ch *Channel) QueueInspect(name string) (wabbit.Queue, error)
  • func (ch *Channel) QueueUnbind(name, route, exchange string, _ wabbit.Option) error
• type Confirmation
  • func (c Confirmation) Ack() bool
  • func (c Confirmation) DeliveryTag() uint64
• type Conn
  • func Dial(uri string) (*Conn, error)
  • func DialConfig(uri string, config amqp.Config) (*Conn, error)
  • func DialTLS(uri string, tlsconfig *tls.Config) (*Conn, error)
  • func (conn *Conn) AutoRedial(outChan chan wabbit.Error, done chan bool)
  • func (conn *Conn) Channel() (wabbit.Channel, error)
  • func (conn *Conn) NotifyClose(c chan wabbit.Error) chan wabbit.Error
• type Delivery
  • func (d *Delivery) Body() []byte
  • func (d *Delivery) ConsumerTag() string
  • func (d *Delivery) ContentType() string
  • func (d *Delivery) DeliveryTag() uint64
  • func (d *Delivery) Headers() wabbit.Option
  • func (d *Delivery) MessageId() string
  • func (d *Delivery) Timestamp() time.Time
• type Publisher
  • func NewPublisher(conn wabbit.Conn, channel wabbit.Channel) (*Publisher, error)
  • func (pb *Publisher) Publish(exc string, route string, message []byte, opt wabbit.Option) error
• type Queue
  • func (q *Queue) Consumers() int
  • func (q *Queue) Messages() int
  • func (q *Queue) Name() string

Types

type Channel

type Channel struct {
	*amqp.Channel
}

Channel is a wrapper channel structure for amqp.Channel

func (*Channel) Confirm

func (ch *Channel) Confirm(noWait bool) error

func (*Channel) Consume

func (ch *Channel) Consume(queue, consumer string, opt wabbit.Option) (<-chan wabbit.Delivery, error)

func (*Channel) ExchangeDeclare

func (ch *Channel) ExchangeDeclare(name, kind string, opt wabbit.Option) error

func (*Channel) ExchangeDeclarePassive

func (ch *Channel) ExchangeDeclarePassive(name, kind string, opt wabbit.Option) error

func (*Channel) NotifyClose

func (ch *Channel) NotifyClose(c chan wabbit.Error) chan wabbit.Error

NotifyClose registers a listener for close events. For more information see: https://godoc.org/github.com/rabbitmq/amqp091-go#Channel.NotifyClose

func (*Channel) NotifyPublish

func (ch *Channel) NotifyPublish(confirm chan wabbit.Confirmation) chan wabbit.Confirmation

func (*Channel) Publish

func (ch *Channel) Publish(exc, route string, msg []byte, opt wabbit.Option) error

func (*Channel) Qos

func (ch *Channel) Qos(prefetchCount, prefetchSize int, global bool) error

Qos controls how many bytes or messages will be handled by channel or connection.

func (*Channel) QueueBind

func (ch *Channel) QueueBind(name, key, exchange string, opt wabbit.Option) error

QueueBind binds the route key to queue

func (*Channel) QueueDeclare

func (ch *Channel) QueueDeclare(name string, opt wabbit.Option) (wabbit.Queue, error)

QueueDeclare declares a new AMQP queue

func (*Channel) QueueDeclarePassive

func (ch *Channel) QueueDeclarePassive(name string, opt wabbit.Option) (wabbit.Queue, error)

QueueDeclarePassive declares an existing AMQP queue

func (*Channel) QueueDelete

func (ch *Channel) QueueDelete(name string, opt wabbit.Option) (int, error)

func (*Channel) QueueInspect

func (ch *Channel) QueueInspect(name string) (wabbit.Queue, error)

func (*Channel) QueueUnbind

func (ch *Channel) QueueUnbind(name, route, exchange string, _ wabbit.Option) error

type Confirmation

type Confirmation struct {
	amqp.Confirmation
}

func (Confirmation) Ack

func (c Confirmation) Ack() bool

func (Confirmation) DeliveryTag

func (c Confirmation) DeliveryTag() uint64

type Conn

type Conn struct {
	*amqp.Connection
	// contains filtered or unexported fields
}

Conn is the amqp connection

func Dial

func Dial(uri string) (*Conn, error)

Dial connects to an AMQP broker, with defaults

func DialConfig

func DialConfig(uri string, config amqp.Config) (*Conn, error)

DialConfig connects to an AMQP broker, with custom config

func DialTLS

func DialTLS(uri string, tlsconfig *tls.Config) (*Conn, error)

DialTLS connects to an AMQP broker, with TLS config

func (*Conn) AutoRedial

func (conn *Conn) AutoRedial(outChan chan wabbit.Error, done chan bool)

AutoRedial manages the automatic redial of connection when unexpected closed. outChan is an unbuffered channel required to receive the errors that results from attempts of reconnect. On successfully reconnected, the true value is sent to done channel

The outChan parameter can receive *amqp.Error for AMQP connection errors or errors.Error for any other net/tcp internal error.

Redial strategy: If the connection is closed in an unexpected way (opposite of conn.Close()), then AutoRedial will try to automatically reconnect waiting for N seconds before each attempt, where N is the number of attempts of reconnecting. If the number of attempts reach 60, it will be zero'ed.

func (*Conn) Channel

func (conn *Conn) Channel() (wabbit.Channel, error)

Channel returns a new channel ready to be used

func (*Conn) NotifyClose

func (conn *Conn) NotifyClose(c chan wabbit.Error) chan wabbit.Error

NotifyClose registers a listener for close events. For more information see: https://godoc.org/github.com/rabbitmq/amqp091-go#Connection.NotifyClose

type Delivery

type Delivery struct {
	*amqp.Delivery
}

func (*Delivery) Body

func (d *Delivery) Body() []byte

func (*Delivery) ConsumerTag

func (d *Delivery) ConsumerTag() string

func (*Delivery) ContentType

func (d *Delivery) ContentType() string

func (*Delivery) DeliveryTag

func (d *Delivery) DeliveryTag() uint64

func (*Delivery) Headers

func (d *Delivery) Headers() wabbit.Option

func (*Delivery) MessageId

func (d *Delivery) MessageId() string

func (*Delivery) Timestamp

func (d *Delivery) Timestamp() time.Time

type Publisher

type Publisher struct {
	// contains filtered or unexported fields
}

func NewPublisher

func NewPublisher(conn wabbit.Conn, channel wabbit.Channel) (*Publisher, error)

func (*Publisher) Publish

func (pb *Publisher) Publish(exc string, route string, message []byte, opt wabbit.Option) error

type Queue

type Queue struct {
	*amqp.Queue
}

Queue is a wrapper for "rabbitmq/amqp091-go".Queue but implementing the wabbit.Queue interface.

func (*Queue) Consumers

func (q *Queue) Consumers() int

Consumers returns the amount of consumers of this queue

func (*Queue) Messages

func (q *Queue) Messages() int

Messages returns the count of messages not awaiting acknowledgment

func (*Queue) Name

func (q *Queue) Name() string

Name of the queue