conn

Documentation

Overview

Package conn adds to https://github.com/streadway/amqp Connection ability to reconnect and some optional parameters.

Index

• Variables
• type Connection
  • func Dial(url string, opts ...ConnectionOption) (*Connection, error)
  • func DialConfig(url string, config amqp.Config, opts ...ConnectionOption) (*Connection, error)
  • func DialTLS(url string, amqps *tls.Config, opts ...ConnectionOption) (*Connection, error)
  • func DialWithDialer(dialer Dialer, opts ...ConnectionOption) (*Connection, error)
  • func Open(conn io.ReadWriteCloser, config amqp.Config, opts ...ConnectionOption) (*Connection, error)
  • func (c *Connection) Channel() (*amqp.Channel, error)
  • func (c *Connection) Close() error
  • func (c *Connection) Connection() *amqp.Connection
  • func (c *Connection) NotifyClose() <-chan Signal
  • func (c *Connection) NotifyConnected(timeout time.Duration) error
• type ConnectionOption
  • func Attempts(n int) ConnectionOption
  • func WithContext(ctx context.Context) ConnectionOption
  • func WithDelay(min, max time.Duration) ConnectionOption
  • func WithDelayBuilder(builder TimeoutBuilder) ConnectionOption
  • func WithLogger(logger logger.Logger) ConnectionOption
• type Connector
  • func ConfigConnector(url string, config amqp.Config, opts ...ConnectionOption) Connector
  • func DefaultConnector(url string, opts ...ConnectionOption) Connector
• type Dialer
• type Notifier
  • func (d *Notifier) Notify()
  • func (d *Notifier) Register(r chan<- Signal)
• type Signal
• type TimeoutBuilder
  • func Backoffer(min, max time.Duration, factor, jitter float64) TimeoutBuilder
  • func CommonTimeoutBuilder(min, max time.Duration) TimeoutBuilder
• type Timeouter
  • func Backoff(min, max time.Duration, factor, jitter float64) Timeouter
  • func CommonTimeouter(min, max time.Duration) Timeouter

Variables

var (
	DeadlineError    = errors.New("the deadline was reached")
	MaxAttemptsError = errors.New("maximum attempts was reached")
	CanceledError    = errors.New("connection was canceled")
	Disconnected     = errors.New("disconnected")
	Connected        = errors.New("connected")
)

Common errors

Types

type Connection

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

Connection is a wrapper of amqp.Connection with reconnection ability.

func Dial

func Dial(url string, opts ...ConnectionOption) (*Connection, error)

Dial wraps amqp.Dial function and adds reconnection ability. Never returns error.

func DialConfig

func DialConfig(url string, config amqp.Config, opts ...ConnectionOption) (*Connection, error)

DialConfig wraps amqp.DialConfig function and adds reconnection ability. Never returns error.

func DialTLS

func DialTLS(url string, amqps *tls.Config, opts ...ConnectionOption) (*Connection, error)

DialTLS wraps amqp.DialTLS function and adds reconnection ability. Never returns error.

func DialWithDialer

func DialWithDialer(dialer Dialer, opts ...ConnectionOption) (*Connection, error)

DialWithDialer wraps any Dialer and adds reconnection ability. Never returns error.

func Open

func Open(conn io.ReadWriteCloser, config amqp.Config, opts ...ConnectionOption) (*Connection, error)

Open wraps amqp.Open function and adds reconnection ability. Never returns error.

func (*Connection) Channel

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

Connection gives direct access to amqp.Connection.

func (*Connection) Close

func (c *Connection) Close() error

func (*Connection) Connection

func (c *Connection) Connection() *amqp.Connection

Connection gives direct access to amqp.Connection.

func (*Connection) NotifyClose

func (c *Connection) NotifyClose() <-chan Signal

NotifyClose notifies user that connection was closed. Channel closes after first notification.

func (*Connection) NotifyConnected

func (c *Connection) NotifyConnected(timeout time.Duration) error

NotifyConnected waits until connection is ready to serve.

type ConnectionOption

type ConnectionOption func(*Connection)

ConnectionOption is a type which represents optional Connection's feature.

func Attempts

func Attempts(n int) ConnectionOption

Attempts sets the maximum attempts to connect/reconnect. When amount rises n, connection stops. When n < 0 Connection tries connect infinitely. -1 by default.

func WithContext

func WithContext(ctx context.Context) ConnectionOption

WithContext allows use power of Context in connection loop. Common use-case: reconnection cancellation.

func WithDelay

func WithDelay(min, max time.Duration) ConnectionOption

Timeout sets delays for connection between attempts.

func WithDelayBuilder

func WithDelayBuilder(builder TimeoutBuilder) ConnectionOption

WithDelayBuilder changes delay mechanism between attempts

func WithLogger

func WithLogger(logger logger.Logger) ConnectionOption

WithLogger sets logger, which notifies about these events:

type Connector

type Connector func() (*Connection, error)

func ConfigConnector

func ConfigConnector(url string, config amqp.Config, opts ...ConnectionOption) Connector

func DefaultConnector

func DefaultConnector(url string, opts ...ConnectionOption) Connector

type Dialer

type Dialer func() (*amqp.Connection, error)

Dialer setups connection to server.

type Notifier

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

Notifier notify receivers when something happen. After notification it closes all channels

func (*Notifier) Notify

func (d *Notifier) Notify()

Send signals to all receivers channels. After notification it closes notified channel. After each notification receiver should register again.

func (*Notifier) Register

func (d *Notifier) Register(r chan<- Signal)

Register registers r as receiver of Notify function.

type Signal

type Signal struct{}

Signal is shortcut for struct{}.

type TimeoutBuilder

type TimeoutBuilder func() Timeouter

func Backoffer

func Backoffer(min, max time.Duration, factor, jitter float64) TimeoutBuilder

func CommonTimeoutBuilder

func CommonTimeoutBuilder(min, max time.Duration) TimeoutBuilder

type Timeouter

type Timeouter interface {
	Wait() // Wait for period of time.
	Inc()  // Increase duration of next Wait call.
}

func Backoff

func Backoff(min, max time.Duration, factor, jitter float64) Timeouter

func CommonTimeouter

func CommonTimeouter(min, max time.Duration) Timeouter