amqp

Documentation

Index

• type Channel
  • 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) 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) QueueDelete(name string, opt wabbit.Option) (int, error)
  • func (ch *Channel) QueueUnbind(name, route, exchange string, _ wabbit.Option) error
• type Conn
  • func Dial(uri string) (*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) DeliveryTag() uint64
• 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) 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) 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) QueueDelete

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

func (*Channel) QueueUnbind

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

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 to AMQP broker

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/streadway/amqp#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) DeliveryTag

func (d *Delivery) DeliveryTag() uint64

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 "streadway/amqp".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