consumer

Jan 26, 2023

Changes in this version

type AMQPChannel

+ Cancel func(consumer string, noWait bool) error

Jan 7, 2023

Changes in this version

type Ready

+ Args amqp.Table

+ AutoAck bool

+ Consumer string

+ DeclareArgs amqp.Table

+ DeclareAutoDelete bool

+ DeclareDurable bool

+ DeclareExclusive bool

+ DeclareNoWait bool

+ DeclareQueue bool

+ Exclusive bool

+ NoLocal bool

+ NoWait bool

+ PrefetchCount int

Apr 23, 2021

Changes in this version

type Option

+ func WithRetryPeriodFunc(durFunc func(retryCount int) time.Duration) Option

Dec 26, 2020

Changes in this version

type AMQPChannel

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

type Option

+ func WithDeclareQueue(name string, durable, autoDelete, exclusive, noWait bool, args amqp.Table) Option

+ func WithQos(prefetchCount int, global bool) Option

+ func WithTmpQueue() Option

+ type State struct

+ Ready *Ready

+ Unready *Unready

+ type Unready struct

+ Err error

Nov 12, 2020

Changes in this version

type AMQPChannel

+ 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 Consumer

+ func (c *Consumer) Notify(readyCh chan Ready, unreadyCh chan error) (ready <-chan Ready, unready <-chan error)

type Option

+ func WithExchange(exchange, routingKey string) Option

+ func WithHandler(h Handler) Option

+ func WithNotify(readyCh chan Ready, unreadyCh chan error) Option

+ func WithQueue(queue string) Option

+ type Ready struct

+ Queue string

Oct 4, 2020

Changes in this version

+ type AMQPChannel interface

+ Close func() error

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

+ NotifyCancel func(c chan string) chan string

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

+ type AMQPConnection interface

+ type Connection struct

+ func NewConnection(amqpConn AMQPConnection, closeCh chan struct{}) *Connection

+ func (c *Connection) AMQPConnection() AMQPConnection

+ func (c *Connection) NotifyClose() chan struct{}

+ type Consumer struct

+ func New(queue string, handler Handler, connCh <-chan *Connection, opts ...Option) (*Consumer, error)

+ func (c *Consumer) Close()

+ func (c *Consumer) NotifyClosed() <-chan struct{}

+ func (c *Consumer) NotifyReady() <-chan struct{}

+ func (c *Consumer) NotifyUnready() <-chan error

+ type DefaultWorker struct

+ Logger logger.Logger

+ func (dw *DefaultWorker) Serve(ctx context.Context, h Handler, msgCh <-chan amqp.Delivery)

+ type Handler interface

+ Handle func(ctx context.Context, msg amqp.Delivery) interface{}

+ func Wrap(handler Handler, middlewares ...Middleware) Handler

+ type HandlerFunc func(ctx context.Context, msg amqp.Delivery) interface

+ func (f HandlerFunc) Handle(ctx context.Context, msg amqp.Delivery) interface{}

+ type Middleware func(next Handler) Handler

+ type Option func(c *Consumer)

+ func WithConsumeArgs(consumer string, autoAck, exclusive, noLocal, noWait bool, args amqp.Table) Option

+ func WithContext(ctx context.Context) Option

+ func WithInitFunc(f func(conn AMQPConnection) (AMQPChannel, error)) Option

+ func WithLogger(l logger.Logger) Option

+ func WithRetryPeriod(dur time.Duration) Option

+ func WithWorker(w Worker) Option

+ type ParallelWorker struct

+ Logger logger.Logger

+ Num int

+ func NewParallelWorker(num int) *ParallelWorker

+ func (pw *ParallelWorker) Serve(ctx context.Context, h Handler, msgCh <-chan amqp.Delivery)

+ type Worker interface

+ Serve func(ctx context.Context, h Handler, msgCh <-chan amqp.Delivery)