infrarabbit

Documentation

Index

• Constants
• func NewBinder(config *ConnectionConfig) (*binder, error)
• type BindConfig
• type ConnectionConfig
  • func (c *ConnectionConfig) Validate() error
• type ConnectionsConfig
  • func (c *ConnectionsConfig) Validate() error
• type Consumer
  • func (c *Consumer) Close() error
  • func (c *Consumer) Consume() chan *Message
• type ConsumerConfig
• type ConsumerMetrics
• type Container
  • func NewContainer() *Container
  • func (cont *Container) AddConnection(name string, cfg *ConnectionConfig) error
  • func (cont *Container) CreateConsumer(consumerCfg *ConsumerConfig) (*Consumer, error)
  • func (cont *Container) CreateProducer(producerCfg *ProducerConfig) (*Producer, error)
• type Kind
• type Message
  • func (m *Message) Ack() error
  • func (m *Message) Body() []byte
  • func (m *Message) IsRedelivered() bool
  • func (m *Message) Nack() error
• type Producer
  • func (p *Producer) Close() error
  • func (p *Producer) Produce(pCtx context.Context, msg *ProducerMessage) error
• type ProducerConfig
• type ProducerMessage

Constants

const PriorityProperty = "x-max-priority"

Functions

func NewBinder

func NewBinder(config *ConnectionConfig) (*binder, error)

Types

type BindConfig

type BindConfig struct {
	Exchange   string // required
	RoutingKey string // required
	Queue      string // required

	// optional
	ExchangeKind       Kind
	ExchangeDurable    bool
	ExchangeAutoDelete bool
	ExchangeInternal   bool
	ExchangeNoWait     bool
	ExchangeArgs       map[string]interface{}
	QueueDurable       bool
	QueueAutoDelete    bool
	QueueExclusive     bool
	QueueNoWait        bool
	QueueArgs          map[string]interface{}
	BindNoWait         bool
	BindArgs           map[string]interface{}
}

type ConnectionConfig

type ConnectionConfig struct {
	Address  string `mapstructure:"address"`
	Username string `mapstructure:"username"`
	Password string `mapstructure:"password"`
	Vhost    string `mapstructure:"vhost"`
}

func (*ConnectionConfig) Validate

func (c *ConnectionConfig) Validate() error

type ConnectionsConfig

type ConnectionsConfig map[string]*ConnectionConfig

func (*ConnectionsConfig) Validate

func (c *ConnectionsConfig) Validate() error

type Consumer

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

func (*Consumer) Close

func (c *Consumer) Close() error

func (*Consumer) Consume

func (c *Consumer) Consume() chan *Message

type ConsumerConfig

type ConsumerConfig struct {
	ConnectionName string
	Queue          string
	QueuePriority  uint8            // optional
	PrefetchCount  int              // optional
	Tag            string           // optional
	Metrics        *ConsumerMetrics // optional
}

type ConsumerMetrics

type ConsumerMetrics struct {
	CheckInterval time.Duration                         // optional
	QueueLength   func(host, queue string, value int64) // optional
	QueueDelay    func(host, queue string, value int64) // optional
}

type Container

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

Container is a simple container for holding named rabbit connections.

func NewContainer

func NewContainer() *Container

func (*Container) AddConnection

func (cont *Container) AddConnection(name string, cfg *ConnectionConfig) error

AddConnection adds a named connection to a container. It's possible to create consumer or producer on created connection later using CreateProducer ot CreateConsumer.

func (*Container) CreateConsumer

func (cont *Container) CreateConsumer(consumerCfg *ConsumerConfig) (*Consumer, error)

CreateConsumer creates a new rabbit consumer by a connection name and subscribes it to a given queue

func (*Container) CreateProducer

func (cont *Container) CreateProducer(producerCfg *ProducerConfig) (*Producer, error)

type Kind

type Kind string

const KindDirect Kind = "direct"

const KindFanOut Kind = "fanout"

const KindHeaders Kind = "headers"

const KindTopic Kind = "topic"

type Message

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

func (*Message) Ack

func (m *Message) Ack() error

func (*Message) Body

func (m *Message) Body() []byte

func (*Message) IsRedelivered

func (m *Message) IsRedelivered() bool

func (*Message) Nack

func (m *Message) Nack() error

type Producer

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

func (*Producer) Close

func (p *Producer) Close() error

func (*Producer) Produce

func (p *Producer) Produce(pCtx context.Context, msg *ProducerMessage) error

type ProducerConfig

type ProducerConfig struct {
	ConnectionName string
	Bindings       []*BindConfig
}

type ProducerMessage

type ProducerMessage struct {
	Body       []byte
	Exchange   string
	RoutingKey string
	Priority   uint8
}