rabbitmq

Documentation

Index

• Constants
• type Bind
• type ConsumerConfig
• type ControlQosConfig
• type Exchange
• type Message
• type ProducerConfig
• type Queue
• type RabbitInterface
  • func New(conf *config.Config) RabbitInterface
• type Rbm_pool
  • func (rbm *Rbm_pool) CloseConnection() error
  • func (rbm *Rbm_pool) CompleteDeclare(cq []Queue, ce []Exchange) []error
  • func (rbm *Rbm_pool) CompleteExchangeDeclare(ce []Exchange) []error
  • func (rbm *Rbm_pool) CompleteQueueDeclare(cq []Queue) []error
  • func (rbm *Rbm_pool) Connect() (RabbitInterface, error)
  • func (rbm *Rbm_pool) Consumer(cc *ConsumerConfig, callback func(msg *amqp.Delivery))
  • func (rbm *Rbm_pool) GetAmqpChannel() *amqp.Channel
  • func (rbm *Rbm_pool) GetAmqpConnection() *amqp.Connection
  • func (rbm *Rbm_pool) GetConnect() *Rbm_pool
  • func (rbm *Rbm_pool) GetConnectStatus() bool
  • func (rbm *Rbm_pool) Producer(ctx context.Context, pc *ProducerConfig, msg *Message) error
  • func (rbm *Rbm_pool) SimpleExchangeDeclare(se Exchange) error
  • func (rbm *Rbm_pool) SimpleQueueDeclare(sq Queue) (queue amqp.Queue, err error)
  • func (rbm *Rbm_pool) StartConsumer(cc *ConsumerConfig, callback func(msg *amqp.Delivery))

Constants

const DEFAULT_MAX_RECONNECT_TIMES = 3 // 3 attempts

Types

type Bind

type Bind struct {
	ExchangeName string
	BindingKey   string
}

type ConsumerConfig

type ConsumerConfig struct {
	Queue            string
	Consumer         string
	AutoAck          bool
	Exclusive        bool
	NoLocal          bool
	NoWait           bool
	Args             amqp.Table
	ControlQosConfig *ControlQosConfig
}

type ControlQosConfig

type ControlQosConfig struct {
	PrefetchCount int
	PrefetchSize  int
	Global        bool
}

type Exchange

type Exchange struct {
	Name       string     // name
	Kind       string     // kind of exchange. ex: 'direct' | 'topic' | 'fanout'
	Durable    bool       // durable
	AutoDelete bool       // delete when unused
	Internal   bool       // internal exchange
	NoWait     bool       // no-wait
	Arguments  amqp.Table // arguments
}

type Message

type Message struct {
	Data        []byte
	ContentType string
}

type ProducerConfig

type ProducerConfig struct {
	Exchange  string
	Key       string
	Mandatory bool
	Immediate bool
}

type Queue

type Queue struct {
	Name       string     // name
	Durable    bool       // durable
	AutoDelete bool       // delete when unused
	Exclusive  bool       // exclusive
	NoWait     bool       // no-wait
	Arguments  amqp.Table // arguments
	Binds      *[]Bind    // bind to exchange and route with queue bind
}

type RabbitInterface

type RabbitInterface interface {
	// Connect creates a new connection and returns RabbitInterface to access functions and error
	Connect() (RabbitInterface, error)
	// GetConnect gets the active connection
	GetConnect() *Rbm_pool
	// GetConnectStatus get the status of connection
	GetConnectStatus() bool
	// CloseConnection closes the active connection
	CloseConnection() error

	// SimpleQueueDeclare used to declare a single Queue into RabbitMQ and returns it or an error
	SimpleQueueDeclare(sq Queue) (queue amqp.Queue, err error)
	// CompleteQueueDeclare used to declare a multiple Queue into RabbitMQ and returns a list of errors if happens.
	//
	// NOTE: If you run this function defining the Bind field contained in Queue, you must have to had defined
	// an Exchange first and then passing it to the field.
	CompleteQueueDeclare(sq []Queue) []error

	// SimpleExchangeDeclare used to declare a single Exchange into RabbitMQ and returns an error if happens
	SimpleExchangeDeclare(se Exchange) error
	// CompleteExchangeDeclare used to declare a multiple Exchange into RabbitMQ and returns a list of errors if happens
	CompleteExchangeDeclare(ce []Exchange) []error

	// CompleteDeclare used to fully declare multiple Queue and Exchange into
	// RabbitMQ and returns a list of errors if happens.
	//
	// NOTE: You can pass empty arrays to this function if not present in your project. If your project
	// doesn't contain binds, just don't set the Bind field contained in Queue struct.
	CompleteDeclare(cq []Queue, ce []Exchange) []error

	// Producer publishes a Message to RabbitMQ following the configuration passed on ProducerConfig
	Producer(ctx context.Context, pc *ProducerConfig, msg *Message) error
	// Consumer consumes a Queue on RabbitMQ following the configuration passed on ConsumerConfig
	Consumer(cc *ConsumerConfig, callback func(msg *amqp.Delivery))
	// StartConsumer starts a consumer routine listening to a Queue of RabbitMQ
	// following the configuration passed on ConsumerConfig.
	//
	// There is a DEFAULT_MAX_RECONNECT_TIMES variable that defines on 3 the number of retries to reconnect to the
	// RabbitMQ service currently running. You can define this number by setting an env variable called
	// SRV_RMQ_MAXX_RECONNECT_TIMES
	StartConsumer(cc *ConsumerConfig, callback func(msg *amqp.Delivery))

	GetAmqpConnection() *amqp.Connection

	GetAmqpChannel() *amqp.Channel
}

func New

func New(conf *config.Config) RabbitInterface

type Rbm_pool

type Rbm_pool struct {
	Channel *amqp.Channel

	MAXX_RECONNECT_TIMES int
	// contains filtered or unexported fields
}

func (*Rbm_pool) CloseConnection

func (rbm *Rbm_pool) CloseConnection() error

func (*Rbm_pool) CompleteDeclare

func (rbm *Rbm_pool) CompleteDeclare(cq []Queue, ce []Exchange) []error

func (*Rbm_pool) CompleteExchangeDeclare

func (rbm *Rbm_pool) CompleteExchangeDeclare(ce []Exchange) []error

func (*Rbm_pool) CompleteQueueDeclare

func (rbm *Rbm_pool) CompleteQueueDeclare(cq []Queue) []error

func (*Rbm_pool) Connect

func (rbm *Rbm_pool) Connect() (RabbitInterface, error)

func (*Rbm_pool) Consumer

func (rbm *Rbm_pool) Consumer(cc *ConsumerConfig, callback func(msg *amqp.Delivery))

func (*Rbm_pool) GetAmqpChannel

func (rbm *Rbm_pool) GetAmqpChannel() *amqp.Channel

func (*Rbm_pool) GetAmqpConnection

func (rbm *Rbm_pool) GetAmqpConnection() *amqp.Connection

func (*Rbm_pool) GetConnect

func (rbm *Rbm_pool) GetConnect() *Rbm_pool

func (*Rbm_pool) GetConnectStatus

func (rbm *Rbm_pool) GetConnectStatus() bool

func (*Rbm_pool) Producer

func (rbm *Rbm_pool) Producer(ctx context.Context, pc *ProducerConfig, msg *Message) error

func (*Rbm_pool) SimpleExchangeDeclare

func (rbm *Rbm_pool) SimpleExchangeDeclare(se Exchange) error

func (*Rbm_pool) SimpleQueueDeclare

func (rbm *Rbm_pool) SimpleQueueDeclare(sq Queue) (queue amqp.Queue, err error)

func (*Rbm_pool) StartConsumer

func (rbm *Rbm_pool) StartConsumer(cc *ConsumerConfig, callback func(msg *amqp.Delivery))