consumer

Documentation

Index

• type AMQPChannel
• type AMQPConnection
• type Connection
  • func NewConnection(amqpConn AMQPConnection, closeCh chan struct{}) *Connection
  • func (c *Connection) AMQPConnection() AMQPConnection
  • func (c *Connection) NotifyClose() chan struct{}
• type Consumer
  • func New(connCh <-chan *Connection, opts ...Option) (*Consumer, error)
  • func (c *Consumer) Close()
  • func (c *Consumer) Notify(stateCh chan State) <-chan State
  • func (c *Consumer) NotifyClosed() <-chan struct{}
• type DefaultWorker
  • func (dw *DefaultWorker) Serve(ctx context.Context, h Handler, msgCh <-chan amqp.Delivery)
• type Handler
  • func Wrap(handler Handler, middlewares ...Middleware) Handler
• type HandlerFunc
  • func (f HandlerFunc) Handle(ctx context.Context, msg amqp.Delivery) interface{}
• type Middleware
• type Option
  • func WithConsumeArgs(consumer string, autoAck, exclusive, noLocal, noWait bool, args amqp.Table) Option
  • func WithContext(ctx context.Context) Option
  • func WithDeclareQueue(name string, durable, autoDelete, exclusive, noWait bool, args amqp.Table) Option
  • func WithExchange(exchange, routingKey string) Option
  • func WithHandler(h Handler) Option
  • func WithInitFunc(f func(conn AMQPConnection) (AMQPChannel, error)) Option
  • func WithLogger(l logger.Logger) Option
  • func WithNotify(stateCh chan State) Option
  • func WithQos(prefetchCount int, global bool) Option
  • func WithQueue(queue string) Option
  • func WithRetryPeriod(dur time.Duration) Option
  • func WithRetryPeriodFunc(durFunc func(retryCount int) time.Duration) Option
  • func WithTmpQueue() Option
  • func WithWorker(w Worker) Option
• type ParallelWorker
  • func NewParallelWorker(num int) *ParallelWorker
  • func (pw *ParallelWorker) Serve(ctx context.Context, h Handler, msgCh <-chan amqp.Delivery)
• type Ready
• type State
• type Unready
• type Worker

Examples

• Wrap

Types

type AMQPChannel

type AMQPChannel interface {
	Consume(queue, consumer string, autoAck, exclusive, noLocal, noWait bool, args amqp.Table) (<-chan amqp.Delivery, error)
	Qos(prefetchCount, prefetchSize int, global bool) error
	NotifyClose(receiver chan *amqp.Error) chan *amqp.Error
	NotifyCancel(c chan string) chan string
	QueueDeclare(name string, durable, autoDelete, exclusive, noWait bool, args amqp.Table) (amqp.Queue, error)
	QueueBind(name, key, exchange string, noWait bool, args amqp.Table) error
	Cancel(consumer string, noWait bool) error
	Close() error
}

type AMQPConnection

type AMQPConnection interface {
}

type Connection

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

func NewConnection

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

func (*Connection) AMQPConnection

func (c *Connection) AMQPConnection() AMQPConnection

func (*Connection) NotifyClose

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

type Consumer

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

func New

func New(
	connCh <-chan *Connection,
	opts ...Option,
) (*Consumer, error)

func (*Consumer) Close

func (c *Consumer) Close()

func (*Consumer) Notify

func (c *Consumer) Notify(stateCh chan State) <-chan State

func (*Consumer) NotifyClosed

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

type DefaultWorker

type DefaultWorker struct {
	Logger logger.Logger
}

func (*DefaultWorker) Serve

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

type Handler

type Handler interface {
	Handle(ctx context.Context, msg amqp.Delivery) interface{}
}

func Wrap

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

Example

package main

import (
	"context"

	"github.com/latifrons/amqpextra/consumer"
	"github.com/latifrons/amqpextra/consumer/middleware"

	amqp "github.com/rabbitmq/amqp091-go"
)

func main() {
	// wrap a handler by some middlewares
	consumer.Wrap(
		consumer.HandlerFunc(func(ctx context.Context, msg amqp.Delivery) interface{} {
			// process message

			msg.Ack(false)

			return nil
		}),
		middleware.HasCorrelationID(),
		middleware.HasReplyTo(),
	)

}

type HandlerFunc

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

func (HandlerFunc) Handle

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

type Middleware

type Middleware func(next Handler) Handler

type Option

type Option func(c *Consumer)

func WithConsumeArgs

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

func WithContext

func WithContext(ctx context.Context) Option

func WithDeclareQueue

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

func WithExchange

func WithExchange(exchange, routingKey string) Option

func WithHandler

func WithHandler(h Handler) Option

func WithInitFunc

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

func WithLogger

func WithLogger(l logger.Logger) Option

func WithNotify

func WithNotify(stateCh chan State) Option

func WithQos

func WithQos(prefetchCount int, global bool) Option

func WithQueue

func WithQueue(queue string) Option

func WithRetryPeriod

func WithRetryPeriod(dur time.Duration) Option

func WithRetryPeriodFunc

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

func WithTmpQueue

func WithTmpQueue() Option

func WithWorker

func WithWorker(w Worker) Option

type ParallelWorker

type ParallelWorker struct {
	Num    int
	Logger logger.Logger
}

func NewParallelWorker

func NewParallelWorker(num int) *ParallelWorker

func (*ParallelWorker) Serve

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

type Ready

type Ready struct {
	Queue string

	PrefetchCount int

	DeclareQueue      bool
	DeclareDurable    bool
	DeclareAutoDelete bool
	DeclareExclusive  bool
	DeclareNoWait     bool
	DeclareArgs       amqp.Table

	Consumer  string
	AutoAck   bool
	Exclusive bool
	NoLocal   bool
	NoWait    bool
	Args      amqp.Table
}

type State

type State struct {
	Unready *Unready
	Ready   *Ready
}

type Unready

type Unready struct {
	Err error
}

type Worker

type Worker interface {
	Serve(ctx context.Context, h Handler, msgCh <-chan amqp.Delivery)
}