infrarabbit

Documentation

Index

• Variables
• 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 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, message []byte, exchange string, routingKey string) error
• type ProducerConfig

Variables

var (
	MetricsNamespace = ""

	ConsumerQueueLength *prometheus.GaugeVec
	ConsumerQueueDelay  *prometheus.GaugeVec
)

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
	PrefetchCount  int
	Tag            string
}

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, message []byte, exchange string, routingKey string) error

type ProducerConfig

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