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(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
  • 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 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
  • func NewParallelWorker(num int) *ParallelWorker
  • func (pw *ParallelWorker) Serve(ctx context.Context, h Handler, msgCh <-chan amqp.Delivery)
• 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)
	NotifyClose(receiver chan *amqp.Error) chan *amqp.Error
	NotifyCancel(c chan string) chan string
	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(
	queue string,
	handler Handler,
	connCh <-chan *Connection,
	opts ...Option,
) (*Consumer, error)

func (*Consumer) Close

func (c *Consumer) Close()

func (*Consumer) NotifyClosed

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

func (*Consumer) NotifyReady

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

func (*Consumer) NotifyUnready

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

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/makasim/amqpextra/consumer"
	"github.com/makasim/amqpextra/consumer/middleware"
	"github.com/streadway/amqp"
)

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 WithInitFunc

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

func WithLogger

func WithLogger(l logger.Logger) Option

func WithRetryPeriod

func WithRetryPeriod(dur time.Duration) 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 Worker

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