rabbitmq

Documentation

Index

• Constants
• func GenerateQueueName(ex string, topic string) string
• type Broker
  • func (b *Broker) Dial(url string) (RBConnection, error)
  • func (b *Broker) DialTLS(url string, conf *tls.Config) (RBConnection, error)
• type ChannelConsumer
• type ChannelCreator
• type ConnectionManager
  • func (m *ConnectionManager) Channel() (RabbitChannel, error)
  • func (m *ConnectionManager) Connect(connectionURL string) (<-chan *amqp.Error, error)
  • func (m *ConnectionManager) Disconnect()
• type Connector
• type Exchange
  • func (e *Exchange) Start() error
  • func (e *Exchange) StartConsuming(topic string, deliveries <-chan amqp.Delivery)
  • func (e *Exchange) Stop()
• type ExchangeFactory
  • func (f *ExchangeFactory) Build() (ExchangeOrganizer, error)
  • func (f *ExchangeFactory) WithChanCreator(creator ChannelCreator) Factory
  • func (f *ExchangeFactory) WithExchange(ex *types.Exchange) Factory
  • func (f *ExchangeFactory) WithInvoker(client types.Invoker) Factory
• type ExchangeHandler
• type ExchangeOrganizer
  • func NewExchange(channel ChannelConsumer, client types.Invoker, definition *types.Exchange) ExchangeOrganizer
• type Factory
  • func NewFactory() Factory
• type Manager
  • func NewConnectionManager(dialer RBDialer) Manager
• type QueueHandler
• type RBConnection
• type RBDialer
  • func NewBroker() RBDialer
• type RabbitChannel
• type Starter
• type Stopper

Constants

const MaxAttempts = 3

MaxAttempts of retries that will be performed

Functions

func GenerateQueueName

func GenerateQueueName(ex string, topic string) string

GenerateQueueName is responsible to generate a unique queue for the connector to use It follows the naming schema OpenFaaS_[EXCHANGE_NAME]_[TOPIC]

Types

type Broker

type Broker struct{}

Broker is a wrapper around the RabbitMQ Client lib, which allows better unit testing. By abstracting away the RabbitMQ raw types, which are struct based.

func (*Broker) Dial

func (b *Broker) Dial(url string) (RBConnection, error)

Dial tries to connect to the providing url, returning either a RBConnection or the received connection error.

func (*Broker) DialTLS

func (b *Broker) DialTLS(url string, conf *tls.Config) (RBConnection, error)

DialTLS tries to connect to the providing url using TLS, returning either a RBConnection or the received connection error.

type ChannelConsumer

type ChannelConsumer interface {
	Consume(queue string, consumer string, autoAck bool, exclusive bool, noLocal bool, noWait bool, args amqp.Table) (<-chan amqp.Delivery, error)
	NotifyClose(c chan *amqp.Error) chan *amqp.Error
	Close() error
}

ChannelConsumer are interacting on channels

type ChannelCreator

type ChannelCreator interface {
	Channel() (RabbitChannel, error)
}

ChannelCreator interface allows the generations of channels

type ConnectionManager

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

ConnectionManager is tasked with managing the connection Rabbit MQ

func (*ConnectionManager) Channel

func (m *ConnectionManager) Channel() (RabbitChannel, error)

Channel creates a new Rabbit MQ channel on the existing connection

func (*ConnectionManager) Connect

func (m *ConnectionManager) Connect(connectionURL string) (<-chan *amqp.Error, error)

Connect uses the provided connection urls and tries up to 3 times to establish a connection. The retries are performed exponentially starting with 2s. It also creates a listener for close notifications.

func (*ConnectionManager) Disconnect

func (m *ConnectionManager) Disconnect()

Disconnect closes the connection and frees up the reference

type Connector

type Connector interface {
	Connect(connectionURL string) (<-chan *amqp.Error, error)
	Disconnect()
}

Connector is a high level interface for connection related methods

type Exchange

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

Exchange contains all of the relevant units to handle communication with an exchange

func (*Exchange) Start

func (e *Exchange) Start() error

Start s consuming deliveries from a unique queue for the specific exchange. Further creating a listener for channel errors

func (*Exchange) StartConsuming

func (e *Exchange) StartConsuming(topic string, deliveries <-chan amqp.Delivery)

StartConsuming will consume deliveries from the provided channel and if the received delivery is for the target topic it will invoke it. If the delivery is not for the correct topic it will reject it so that the delivery is returned to the exchange. Retries are exponential and up to 3 times.

func (*Exchange) Stop

func (e *Exchange) Stop()

Stop s consuming messages

type ExchangeFactory

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

ExchangeFactory keeps tracks of all the build options provided to it during construction

func (*ExchangeFactory) Build

func (f *ExchangeFactory) Build() (ExchangeOrganizer, error)

Build uses the set values and builds a new exchange from them

func (*ExchangeFactory) WithChanCreator

func (f *ExchangeFactory) WithChanCreator(creator ChannelCreator) Factory

WithChanCreator sets the channel creator that will be used

func (*ExchangeFactory) WithExchange

func (f *ExchangeFactory) WithExchange(ex *types.Exchange) Factory

WithExchange sets the exchange definition and further ensures that the correct type is used

func (*ExchangeFactory) WithInvoker

func (f *ExchangeFactory) WithInvoker(client types.Invoker) Factory

WithInvoker sets the invoker which will interact with OpenFaaS

type ExchangeHandler

type ExchangeHandler interface {
	ExchangeDeclare(name, kind string, durable, autoDelete, internal, noWait bool, args amqp.Table) error
}

ExchangeHandler offers a interface for the decleration of an exchange or the validation against existing exchanges on the RabbitMQ cluster

type ExchangeOrganizer

type ExchangeOrganizer interface {
	Starter
	Stopper
}

ExchangeOrganizer combines the ability to start & stop exchanges

func NewExchange

func NewExchange(channel ChannelConsumer, client types.Invoker, definition *types.Exchange) ExchangeOrganizer

NewExchange creates a new exchange instance using the provided parameter

type Factory

type Factory interface {
	WithInvoker(client types.Invoker) Factory
	WithChanCreator(creator ChannelCreator) Factory
	WithExchange(ex *types.Exchange) Factory
	Build() (ExchangeOrganizer, error)
}

Factory for building a Exchange

func NewFactory

func NewFactory() Factory

NewFactory creates a new instance with no defaults set.

type Manager

type Manager interface {
	Connector
	ChannelCreator
}

Manager is a interface that combines the relevant methods to connect to Rabbit MQ And create a new channel on an existing connection.

func NewConnectionManager

func NewConnectionManager(dialer RBDialer) Manager

NewConnectionManager creates a new instance using the provided dialer

type QueueHandler

type QueueHandler interface {
	QueueDeclare(name string, durable, autoDelete, exclusive, noWait bool, args amqp.Table) (amqp.Queue, error)
	QueueBind(name, key, exchange string, noWait bool, args amqp.Table) error
}

QueueHandler offers a interface for the decleration & binding of an queues. Further it allows the validation against existing queues on the RabbitMQ cluster

type RBConnection

type RBConnection interface {
	NotifyClose(receiver chan *amqp.Error) chan *amqp.Error
	Close() error
	Channel() (*amqp.Channel, error)
}

RBConnection is a abstraction of a RabbitMQ Connection

type RBDialer

type RBDialer interface {
	Dial(url string) (RBConnection, error)
	DialTLS(url string, conf *tls.Config) (RBConnection, error)
}

RBDialer is a abstraction of the RabbitMQ Dial methods

func NewBroker

func NewBroker() RBDialer

NewBroker generates a new wrapper around the RabbitMQ Client lib

type RabbitChannel

type RabbitChannel interface {
	ExchangeHandler
	QueueHandler
	ChannelConsumer
}

RabbitChannel is a abstraction of a RabbitMQ Channel

type Starter

type Starter interface {
	Start() error
}

Starter defines something that can be started

type Stopper

type Stopper interface {
	Stop()
}

Stopper defines something that can be stopped