mq

Documentation

Overview

Package mq provides an abstraction for handling amqp connection logic

Index

• type AMQPConnector
  • func NewAMQPConnector(cfg *RabbitMQConfig, logger *log.Logger) *AMQPConnector
  • func (f AMQPConnector) Channel() (*amqp.Channel, error)
  • func (f AMQPConnector) Consume(amqpChannel *amqp.Channel, handlers map[string]AMQPHandler) error
• type AMQPConsumer
  • func NewAMQPConsumer(h AMQPHandler, ch *amqp.Channel, logger *log.Logger, queue, tag string) *AMQPConsumer
  • func (c *AMQPConsumer) Consume() error
• type AMQPHandler
• type AMQPHandlerFunc
  • func (f AMQPHandlerFunc) HandleDelivery(d *amqp.Delivery) error
• type AMQPPublisher
  • func NewAMQPPublisher(amqpChannel *amqp.Channel, exchange string) *AMQPPublisher
  • func (p *AMQPPublisher) Publish(b []byte, headers amqp.Table, routingKey string) error
• type Connector
• type Consumer
• type Publisher
• type RabbitMQConfig
• type RabbitMQConnection
  • func (c *RabbitMQConnection) String() string
• type RabbitMQConsumer

Types

type AMQPConnector

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

AMQPConnector implements Connector interface

func NewAMQPConnector

func NewAMQPConnector(cfg *RabbitMQConfig, logger *log.Logger) *AMQPConnector

NewAMQPConnector returns a new instance of AMQPConnector

func (AMQPConnector) Channel

func (f AMQPConnector) Channel() (*amqp.Channel, error)

func (AMQPConnector) Consume

func (f AMQPConnector) Consume(amqpChannel *amqp.Channel, handlers map[string]AMQPHandler) error

Consume receives a list of AMQPHandler implementations and create the instances of the AMQPConsumer using the amqpChannel

type AMQPConsumer

type AMQPConsumer struct {
	Handler     AMQPHandler
	AMQPChannel *amqp.Channel
	Queue       string
	Tag         string
	// contains filtered or unexported fields
}

AMQPConsumer implements Consumer interface

func NewAMQPConsumer

func NewAMQPConsumer(h AMQPHandler, ch *amqp.Channel, logger *log.Logger, queue, tag string) *AMQPConsumer

NewAMQPConsumer inits and returns a pointer to a new AMQPConsumer instance

func (*AMQPConsumer) Consume

func (c *AMQPConsumer) Consume() error

Consume processes amqp messages

type AMQPHandler

type AMQPHandler interface {
	HandleDelivery(*amqp.Delivery) error
}

AMQPHanlder describes logic for processing an message from an amqp server

type AMQPHandlerFunc

type AMQPHandlerFunc func(*amqp.Delivery) error

AMQPHandlerFunc implements AMQPHandler interface

func (AMQPHandlerFunc) HandleDelivery

func (f AMQPHandlerFunc) HandleDelivery(d *amqp.Delivery) error

HandleDelivery calls f(d)

type AMQPPublisher

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

AMQPPublisher implements Publisher interface

func NewAMQPPublisher

func NewAMQPPublisher(amqpChannel *amqp.Channel, exchange string) *AMQPPublisher

NewAMQPPublisher inits and returns an AMQPublisher

func (*AMQPPublisher) Publish

func (p *AMQPPublisher) Publish(b []byte, headers amqp.Table, routingKey string) error

Publish sends a message to the amqp server

type Connector

type Connector interface {
	Channel() (*amqp.Channel, error)
	Consume(*amqp.Channel, map[string]AMQPHandler) error
}

Connector describes amqp dial logic

type Consumer

type Consumer interface {
	Consume() error
}

Consumer interface desribes logic for an amqp consumer

type Publisher

type Publisher interface {
	Publish(b []byte, headers amqp.Table, routingKey string) error
}

Publisher describes logic for publishing messages to the amqp server

type RabbitMQConfig

type RabbitMQConfig struct {
	Connection RabbitMQConnection `json:"connection"`
	Consumers  []RabbitMQConsumer `json:"consumers"`
}

RabbitMQConfig holds rabbitmq server configs

type RabbitMQConnection

type RabbitMQConnection struct {
	User            string `json:"user"`
	Pass            string `json:"pass"`
	Host            string `json:"host"`
	Port            int    `json:"port"`
	Attempts        int    `json:"attempts"`
	ErrLogEnable    bool   `json:"error_log_enable"`
	ReturnLogEnable bool   `json:"return_log_enable"`
	PrefetchCount   int    `json:"prefetch_count"`
}

RabbitMQConnection desribes a single rabbitmq server configuration

func (*RabbitMQConnection) String

func (c *RabbitMQConnection) String() string

String returns a rabbitmq server connection string

type RabbitMQConsumer

type RabbitMQConsumer struct {
	ID      string `json:"id"`
	Queue   string `json:"queue"`
	Workers int    `json:"workers"`
}

RabbitMQConsumer desribes a rabbitmq consumer config