rabbitmq

Nov 24, 2023

Changes in this version

type ConsumerConf

+ Logger log.Logger

type ProducerConf

+ Logger log.Logger

Apr 27, 2023

Changes in this version

+ type Msg struct

+ Body []byte

+ Delay time.Duration

+ Key string

Apr 25, 2023

Changes in this version

+ const XDelayedMessage

+ var DelayTable = amqp.Table

Feb 14, 2023

Changes in this version

type Consumer

+ func (c *Consumer) Init() error

type Producer

+ func (p *Producer) Init() error

type RabbitConsumer

+ Close func() error

+ Init func() error

type RabbitProducer

+ Close func() error

+ Init func() error

Nov 30, 2022

Changes in this version

+ type RabbitConsumer interface

+ Consume func(handler MsgHandler)

+ type RabbitProducer interface

+ SendMsg func(ctx context.Context, key string, body []byte) error

Nov 30, 2022

Changes in this version

+ const DefaultBatchPullCount

type Consumer

+ func (c *Consumer) Setup() error

+ func (c *Consumer) Watch()

type ConsumerConf

+ RoutingKeys []string

type Producer

+ func (p *Producer) Setup() error

+ func (p *Producer) Watch()

Jul 24, 2022

Changes in this version

+ var CancelChannelErr = errors.New("consumer cancel failed")

+ var CloseConnErr = errors.New("amqp connection close error")

+ var EmptyRoutingKey = errors.New("empty routing key")

+ func BindExchangeQueue(ch *amqp.Channel, queueName, exchangeName, routingKey string, broadcast bool) error

+ func CloseChannel(ch *amqp.Channel) error

+ func Connect(url string) (*amqp.Connection, error)

+ func ExchangeDeclare(ch *amqp.Channel, exchange Exchange) error

+ func GetChannel(conn *amqp.Connection) (*amqp.Channel, error)

+ func QueueDeclare(ch *amqp.Channel, queue Queue) error

type Consumer

+ func NewConsumer(conf ConsumerConf) *Consumer

+ func (c *Consumer) Close() error

+ func (c *Consumer) ReConnect()

+ func (c *Consumer) Start() error

+ type ConsumerConf struct

+ Args amqp.Table

+ AutoAck bool

+ BatchPullCount int

+ ConsumerTag string

+ Exchange Exchange

+ Exclusive bool

+ NoLocal bool

+ NoWait bool

+ Queue Queue

+ RoutingKey string

+ Url string

+ type MsgHandler func(delivery amqp.Delivery)

+ type Producer struct

+ func NewProducer(conf ProducerConf) *Producer

+ func (c *Producer) Close() error

+ func (p *Producer) ReConnect()

+ func (p *Producer) SendMsg(body []byte) error

+ func (p *Producer) Start() error

+ type ProducerConf struct

+ Exchange Exchange

+ Queue Queue

+ RoutingKey string

+ Url string

May 24, 2022

Changes in this version

type RabbitMQ

+ NoLocal bool

May 22, 2022

Changes in this version

+ const DefaultPullBatchSize

+ const Direct

+ const Fanout

+ const Headers

+ const Topic

+ var EmptyConnection = errors.New("empty connection get channel")

+ var EmptyExchange = errors.New("empty exchange")

+ var EmptyQueue = errors.New("empty queue")

+ var EmptyRoutingKeys = errors.New("empty routing key")

+ var ExecuteStartBeforeRecvMsg = errors.New("execute start function before receive msg")

+ var ExecuteStartBeforeSendMsg = errors.New("execute start function before send msg")

+ func DefaultFailCallback(body []byte) error

+ func DefaultSuccessCallback(body []byte) error

+ type ConsumeCallback func(msgId string, body []byte, err error) error

+ type Consumer interface

+ Consume func([]byte) error

+ ConsumeFailConsumeCallback func(msgId string, body []byte, err error)

+ ConsumeSuccessCallback func(msgId string, body []byte)

+ type Exchange struct

+ Arguments map[string]interface{}

+ AutoDeleted bool

+ ExType string

+ Internal bool

+ Name string

+ NoWait bool

+ Persistence bool

+ func (e *Exchange) CheckAndSet() error

+ type Queue struct

+ Arguments map[string]interface{}

+ AutoDeleted bool

+ Exclusive bool

+ Name string

+ NoWait bool

+ Persistence bool

+ func (q *Queue) CheckAndSet() error

+ type RabbitMQ struct

+ Args map[string]interface{}

+ AutoAck bool

+ ConsumeSuccessCallback func([]byte, error) error

+ Exchange *Exchange

+ Exclusive bool

+ NoWait bool

+ PullBatchSize int

+ Queue *Queue

+ RoutingKey string

+ SendFail SendCallback

+ SendSuccess SendCallback

+ SendTeardown SendCallback

+ Url string

+ func (r *RabbitMQ) BindExchangeQueue(ch *amqp.Channel) error

+ func (r *RabbitMQ) Close() error

+ func (r *RabbitMQ) CloseChannel() error

+ func (r *RabbitMQ) Connect() error

+ func (r *RabbitMQ) ConsumeMsg(consumer Consumer)

+ func (r *RabbitMQ) ExchangeDeclare(ch *amqp.Channel) error

+ func (r *RabbitMQ) OpenChannel() error

+ func (r *RabbitMQ) QueueDeclare(ch *amqp.Channel) error

+ func (r *RabbitMQ) RecvMsg() (<-chan amqp.Delivery, error)

+ func (r *RabbitMQ) SendMsg(body []byte) (bool, error)

+ func (r *RabbitMQ) Start() error

+ type SendCallback func(body []byte) error