rabbitmq

Documentation

Index

• Variables
• type Exchange
• type IRabbitMQMessagingClient
• type QueueDetails
• type Queues
  • func InitiateQueues(queueToExchangeMap map[string]Exchange) Queues
  • func ParseAndCreateQueueReference(consumingQueues, producingQueues string, ...) Queues
  • func ParseAndCreateQueueReferences(consumingQueues, producingQueues []string, ...) Queues
• type RabbitMQClient
  • func New(url string, queues Queues, logger *zap.Logger) (*RabbitMQClient, error)
  • func (m *RabbitMQClient) Close()
  • func (m *RabbitMQClient) PublishOnQueue(ctx context.Context, body []byte, exchange string) error
  • func (m *RabbitMQClient) SubscribeToQueue(consumerExchangeName string, callback func(amqp.Delivery) error) error

Variables

var DefaultConnectionString = "amqp://guest:guest@localhost:5672"

Types

type Exchange

type Exchange struct {
	ExchangeName string
	ExchangeType string
	Durable      bool
	AutoDelete   bool
	Internal     bool
	NoWait       bool
	Args         *amqp.Table
}

type IRabbitMQMessagingClient

type IRabbitMQMessagingClient interface {
	PublishOnQueue(ctx context.Context, body []byte, exchange string) error
	SubscribeToQueue(consumerExchangeName string, handlerFunc func(amqp.Delivery) error) error
	Close()
}

IMessagingClient defines an interface for connection and consuming messages

type QueueDetails

type QueueDetails struct {
	QueueName    string
	ExchangeName string
	Durable      bool
	AutoDelete   bool
	Exclusive    bool
	NoWait       bool
	Args         *amqp.Table
}

type Queues

type Queues struct {
	QueueDetails []QueueDetails
	Exchanges    []Exchange
}

func InitiateQueues

func InitiateQueues(queueToExchangeMap map[string]Exchange) Queues

InitiateQueues takes as input a queue to exchange mapping an returns reference to a queue

func ParseAndCreateQueueReference

func ParseAndCreateQueueReference(consumingQueues, producingQueues string, numProducerQueues, numConsumerQueues int, logger *zap.Logger) Queues

func ParseAndCreateQueueReferences

func ParseAndCreateQueueReferences(consumingQueues, producingQueues []string, numProducerQueues, numConsumerQueues int, logger *zap.Logger) Queues

type RabbitMQClient

type RabbitMQClient struct {
	Conn          *amqp.Connection
	Chann         *amqp.Channel
	Logger        *zap.Logger
	QueueBindings map[string]*amqp.Queue
}

func New

func New(url string, queues Queues, logger *zap.Logger) (*RabbitMQClient, error)

New creates an amqp broker connection and binds a set of input queues to each specific exchange Note all queues that the services writes to and consumes must be present and passed as input parameters

func (*RabbitMQClient) Close

func (m *RabbitMQClient) Close()

Close closes the connection to the AMQP-broker, if available.

func (*RabbitMQClient) PublishOnQueue

func (m *RabbitMQClient) PublishOnQueue(ctx context.Context, body []byte, exchange string) error

PublishOnQueue publishes a message on a queue based on exchange name

func (*RabbitMQClient) SubscribeToQueue

func (m *RabbitMQClient) SubscribeToQueue(consumerExchangeName string, callback func(amqp.Delivery) error) error

SubscribeToQueue subscribes a callback to a queue tied to an exchange