rabbitmq

Documentation

Overview

NewRabbitMQ returns a new initialized connection struct. It will manage the active connection in the background. Connection should be closed in order to shut it down gracefully.

func example() {
	user := "guest"
	pass := "guest"
	host := "localhost"
	port := "5672"
	consumer := "user-service" //  Unique name for each consumer used to sign messages.

	// You can specify your own config or use DefaultConfig() instead.
	config := Config{
		QueueSize:         100,
		MaxWorkers:        50,
		ReconnectInterval: time.Second * 2,
		MaxRequests:       5,
		ClearInterval:     time.Second * 5,
		ClosedTimeout:     time.Second * 5,
	}

	// Logger and tracer are optional.
	rabbit := rabbitmq.NewRabbitMQ(consumer, user, pass, host, port, config, WithLogger(customLogger))
	defer rabbit.Close()
}

Index

• Variables
• type Config
  • func DefaultConfig() Config
• type ContentType
• type Message
• type Option
  • func WithLogger(logger logging.Logger) Option
  • func WithTracer(tracer trace.Tracer) Option
• type RabbitMQ
  • func NewRabbitMQ(consumer, user, pass, host, port string, config Config, opts ...Option) *RabbitMQ
  • func (mq *RabbitMQ) Close() error
  • func (mq *RabbitMQ) Consume(ctx context.Context, command string, route Route) (_ <-chan Message, err error)
  • func (mq *RabbitMQ) Enqueue(msg Message) error
  • func (mq *RabbitMQ) Publish(ctx context.Context, msg Message) (err error)
• type Route

Variables

var ErrFullQueue = errors.New("publish queue is full")

Types

type Config

type Config struct {
	QueueSize         int           // Max number of messages internally queued for publishing.
	MaxWorkers        int           // Max number of concurrent workers per operation type.
	ReconnectInterval time.Duration // Time between reconnect attempts.

	// Settings for the internal circuit breaker.
	MaxRequests   uint32        // Number of requests allowed to half-open state.
	ClearInterval time.Duration // Time after which failed calls count is cleared.
	ClosedTimeout time.Duration // Time after which closed state becomes half-open.
}

func DefaultConfig

func DefaultConfig() Config

type ContentType

type ContentType string

const (
	ContentTypeJson ContentType = "application/json"
	ContentTypeText ContentType = "text/plain"
)

type Message

type Message struct {
	Route

	Body        []byte
	ContentType ContentType
	Timestamp   time.Time
	Headers     map[string]string
}

type Option

type Option interface {
	// contains filtered or unexported methods
}

func WithLogger

func WithLogger(logger logging.Logger) Option

func WithTracer

func WithTracer(tracer trace.Tracer) Option

type RabbitMQ

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

func NewRabbitMQ

func NewRabbitMQ(consumer, user, pass, host, port string, config Config, opts ...Option) *RabbitMQ

NewRabbitMQ returns a new initialized connection struct. It will manage the active connection in the background. Connection should be closed in order to shut it down gracefully.

func example() {
	user := "guest"
	pass := "guest"
	host := "localhost"
	port := "5672"
	consumer := "user-service" //  Unique name for each consumer used to sign messages.

	var customLogger Logger

	// You can specify your own config or use rabbitmq.DefaultConfig() instead.
	config := Config{
		QueueSize:         100,
		MaxWorkers:        50,
		ReconnectInterval: time.Second * 2,
		MaxRequests:       5,
		ClearInterval:     time.Second * 5,
		ClosedTimeout:     time.Second * 5,
	}

	// Logger and tracer are optional.
	rabbit := rabbitmq.NewRabbitMQ(consumer, user, pass, host, port, config, WithLogger(customLogger))
	defer rabbit.Close()
}

func (*RabbitMQ) Close

func (mq *RabbitMQ) Close() error

Close closes active connection gracefully.

func (*RabbitMQ) Consume

func (mq *RabbitMQ) Consume(ctx context.Context, command string, route Route) (_ <-chan Message, err error)

func (*RabbitMQ) Enqueue

func (mq *RabbitMQ) Enqueue(msg Message) error

Enqueue appends a message to the publishQueue and returns a non-nil error if the queue is full.

func (*RabbitMQ) Publish

func (mq *RabbitMQ) Publish(ctx context.Context, msg Message) (err error)

type Route

type Route struct {
	ExchangeName string
	ExchangeType string
	RoutingKey   string
}