harego

May 25, 2023

Changes in this version

type ConfigFunc

+ func DeprecatedLogger(l logger) ConfigFunc

+ func WithLogrus(l logrus.FieldLogger) ConfigFunc

+ func WithPanicHandler(h PanicHandler) ConfigFunc

+ type PanicHandler func(msg *amqp.Delivery, r any) (a AckType, delay time.Duration)

Feb 21, 2023

Changes in this version

+ const DeliveryModePersistent

+ const DeliveryModeTransient

+ var ErrAlreadyConfigured = errors.New("client is already configured")

+ var ErrClosed = errors.New("exchange is already closed")

+ var ErrInput = errors.New("invalid input")

+ var ErrNilHnadler = errors.New("handler can not be nil")

+ func AutoAck(c *config)

+ func AutoDelete(c *config)

+ func ExclusiveQueue(c *config)

+ func Internal(c *config)

+ func NoWait(c *config)

+ func NotDurable(c *config)

+ type AckType int

+ const AckTypeAck

+ const AckTypeNack

+ const AckTypeReject

+ const AckTypeRequeue

+ func (a AckType) IsValid() bool

+ func (i AckType) String() string

+ type Channel interface

+ Close func() error

+ Consume func(queue, consumer string, autoAck, exclusive, noLocal, noWait bool, ...) (<-chan amqp.Delivery, error)

+ ExchangeDeclare func(name, kind string, durable, autoDelete, internal, noWait bool, args amqp.Table) error

+ NotifyClose func(receiver chan *amqp.Error) chan *amqp.Error

+ Publish func(exchange, key string, mandatory, immediate bool, msg amqp.Publishing) error

+ Qos func(prefetchCount, prefetchSize int, global bool) error

+ QueueBind func(name, key, exchange string, noWait bool, args amqp.Table) error

+ QueueDeclare func(name string, durable, autoDelete, exclusive, noWait bool, args amqp.Table) (amqp.Queue, error)

+ type ConfigFunc func(*config)

+ func Buffer(n int) ConfigFunc

+ func ConsumerName(name string) ConfigFunc

+ func Context(ctx context.Context) ConfigFunc

+ func ExchangeName(name string) ConfigFunc

+ func Logger(l logger) ConfigFunc

+ func PrefetchCount(i int) ConfigFunc

+ func PrefetchSize(i int) ConfigFunc

+ func QueueArgs(args amqp.Table) ConfigFunc

+ func QueueName(name string) ConfigFunc

+ func RetryDelay(d time.Duration) ConfigFunc

+ func RoutingKey(key string) ConfigFunc

+ func WithDeliveryMode(mode DeliveryMode) ConfigFunc

+ func WithExchangeType(t ExchangeType) ConfigFunc

+ func Workers(n int) ConfigFunc

+ type Connector func() (RabbitMQ, error)

+ func AMQPConnector(r *amqp.Connection) Connector

+ func URLConnector(url string) Connector

+ type Consumer struct

+ func NewConsumer(connector Connector, conf ...ConfigFunc) (*Consumer, error)

+ func (c *Consumer) Close() error

+ func (c *Consumer) Consume(ctx context.Context, handler HandlerFunc) error

+ type DeliveryMode uint8

+ func (d DeliveryMode) IsValid() bool

+ func (i DeliveryMode) String() string

+ type ExchangeType int

+ const ExchangeTypeDirect

+ const ExchangeTypeFanout

+ const ExchangeTypeHeaders

+ const ExchangeTypeTopic

+ func (e ExchangeType) IsValid() bool

+ func (e ExchangeType) String() string

+ type HandlerFunc func(msg *amqp.Delivery) (a AckType, delay time.Duration)

+ type Publisher struct

+ func NewPublisher(connector Connector, conf ...ConfigFunc) (*Publisher, error)

+ func (p *Publisher) Close() error

+ func (p *Publisher) Publish(msg *amqp.Publishing) error

+ type RabbitMQ interface

+ Channel func() (Channel, error)

+ Close func() error