rabbit

Documentation

Index

• func DeclareExchange(ch *amqp.Channel, topic string) error
• func DeclareQueue(ch *amqp.Channel, qName string) (amqp.Queue, error)
• func NewChannel(ch *amqp.Channel, qName string) (<-chan amqp.Delivery, error)
• type ChannelProvider
  • func NewChannelProvider() (*ChannelProvider, error)
  • func (pr *ChannelProvider) Channel() (*amqp.Channel, error)
  • func (pr *ChannelProvider) Close()
  • func (pr *ChannelProvider) Healthy() error
  • func (pr *ChannelProvider) QueueName(name string) string
  • func (pr *ChannelProvider) RunOnChannelWithRetry(f runOnChannelFunc) error
• type Publisher
  • func NewPublisher(provider *ChannelProvider) *Publisher
  • func (sender *Publisher) Publish(id string, topic string) error
• type Sender
  • func NewSender(provider *ChannelProvider) *Sender
  • func (sender *Sender) Send(message messages.Message, queue string, replyQueue string) error
  • func (sender *Sender) SendWithCorr(message messages.Message, queue string, replyQueue string, corrID string) error

Functions

func DeclareExchange

func DeclareExchange(ch *amqp.Channel, topic string) error

DeclareExchange creates exchange to publish events

func DeclareQueue

func DeclareQueue(ch *amqp.Channel, qName string) (amqp.Queue, error)

DeclareQueue decrares durable queue

func NewChannel

func NewChannel(ch *amqp.Channel, qName string) (<-chan amqp.Delivery, error)

NewChannel creates channel to listen from rabbit with auto ack = false

Types

type ChannelProvider

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

ChannelProvider provider amqp channel

func NewChannelProvider

func NewChannelProvider() (*ChannelProvider, error)

NewChannelProvider initializes channel provider

func (*ChannelProvider) Channel

func (pr *ChannelProvider) Channel() (*amqp.Channel, error)

Channel return cached channel or tries to connect to rabbit broker

func (*ChannelProvider) Close

func (pr *ChannelProvider) Close()

Close finalizes ChannelProvider

func (*ChannelProvider) Healthy

func (pr *ChannelProvider) Healthy() error

Healthy checks if rabbit channel is open

func (*ChannelProvider) QueueName

func (pr *ChannelProvider) QueueName(name string) string

QueueName return queue name for channel, may append prefix

func (*ChannelProvider) RunOnChannelWithRetry

func (pr *ChannelProvider) RunOnChannelWithRetry(f runOnChannelFunc) error

RunOnChannelWithRetry invokes method on channel with retry

type Publisher

type Publisher struct {
	ChannelProvider *ChannelProvider
}

Publisher publish events to rabbit mq broker

func NewPublisher

func NewPublisher(provider *ChannelProvider) *Publisher

NewPublisher initializes rabbit publisher

func (*Publisher) Publish

func (sender *Publisher) Publish(id string, topic string) error

Publish publish the message

type Sender

type Sender struct {
	ChannelProvider *ChannelProvider
}

Sender performs messages sending using rabbit mq broker

func NewSender

func NewSender(provider *ChannelProvider) *Sender

NewSender initializes rabbit sender

func (*Sender) Send

func (sender *Sender) Send(message messages.Message, queue string, replyQueue string) error

Send sends the message

func (*Sender) SendWithCorr

func (sender *Sender) SendWithCorr(message messages.Message, queue string, replyQueue string, corrID string) error

SendWithCorr sends the message with correlationID