rabbit

README

RabbitMQ Provider

This package provides a RabbitMQ implementation of the mq.Provider interface.

Features

• ✅ Publish
• ✅ Subscribe
• ✅ Reconnect
• ✅ OpenTelemetry
• ✅ Logging

Documentation

Index

• type Channel
  • func (ch *Channel) Close() error
  • func (ch *Channel) Consume(ctx context.Context, queue, consumer string, ...) (<-chan amqp.Delivery, error)
  • func (ch *Channel) IsClosed() bool
• type Config
• type Connection
  • func (c *Connection) Channel() (*Channel, error)
• type MQ
  • func New(log logger.Logger) *MQ
  • func (mq *MQ) Check(_ context.Context) error
  • func (mq *MQ) Dial() error
  • func (mq *MQ) Init(ctx context.Context, log logger.Logger) error
  • func (mq *MQ) Publish(ctx context.Context, target string, routingKey, payload []byte) error
  • func (mq *MQ) Subscribe(ctx context.Context, target string, message query.Response) error
  • func (mq *MQ) UnSubscribe(target string) error

Types

type Channel

type Channel struct {
	*amqp.Channel
	// contains filtered or unexported fields
}

Channel amqp.Channel wapper

func (*Channel) Close

func (ch *Channel) Close() error

Close ensure closed flag set

func (*Channel) Consume

func (ch *Channel) Consume(ctx context.Context, queue, consumer string, autoAck, exclusive, noLocal, noWait bool, args amqp.Table) (<-chan amqp.Delivery, error)

Consume wrap amqp.Channel.Consume, the returned delivery will end only when channel closed by developer

func (*Channel) IsClosed

func (ch *Channel) IsClosed() bool

IsClosed indicate closed by developer

type Config

type Config struct {
	URI           string
	ReconnectTime int
}

type Connection

type Connection struct {
	*amqp.Connection
	// contains filtered or unexported fields
}

Connection amqp.Connection wrapper

func (*Connection) Channel

func (c *Connection) Channel() (*Channel, error)

Channel wraps amqp.Connection.Channel to provide an auto-reconnecting channel.

type MQ

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

func New

func New(log logger.Logger) *MQ

func (*MQ) Check

func (mq *MQ) Check(_ context.Context) error

Check verifies the connection status.

func (*MQ) Dial

func (mq *MQ) Dial() error

Dial wraps amqp.Dial to establish a connection and set up automatic reconnection in case the connection is lost.

func (*MQ) Init

func (mq *MQ) Init(ctx context.Context, log logger.Logger) error

Init initializes the RabbitMQ connection and sets up the channel. It also sets up a graceful shutdown mechanism to close the connection and channel when the context is done.

func (*MQ) Publish

func (mq *MQ) Publish(ctx context.Context, target string, routingKey, payload []byte) error

func (*MQ) Subscribe

func (mq *MQ) Subscribe(ctx context.Context, target string, message query.Response) error

func (*MQ) UnSubscribe

func (mq *MQ) UnSubscribe(target string) error