rabbit

Documentation

Index

• Constants
• func BuildRabbitQueueConsumer(delivery <-chan amqp.Delivery, handleFunc HandleFunc, consumerName string, ...) (*rabbitQueueConsumer, cerror.AppError)
• func ExchangeEnableAutoDelete(o *ExchangeDeclareOptions)
• func ExchangeEnableDurable(o *ExchangeDeclareOptions)
• func ExchangeEnableInternal(o *ExchangeDeclareOptions)
• func ExchangeEnableNoWait(o *ExchangeDeclareOptions)
• func QueueEnableAutoDelete(o *QueueDeclareOptions)
• func QueueEnableDurable(o *QueueDeclareOptions)
• func QueueEnableExclusive(o *QueueDeclareOptions)
• func QueueEnableNoWait(o *QueueDeclareOptions)
• type ExchangeDeclareOption
• type ExchangeDeclareOptions
• type HandleFunc
• type QueueDeclareOption
• type QueueDeclareOptions
• type RabbitBroker
  • func BuildRabbit(config *conf.RabbitConfig, errChan chan<- cerror.AppError) (*RabbitBroker, cerror.AppError)
  • func (l *RabbitBroker) Consume(queueName string, consumerName string) (<-chan amqp.Delivery, cerror.AppError)
  • func (l *RabbitBroker) ExchangeBind(destination string, key string, source string, noWait bool, ...) cerror.AppError
  • func (l *RabbitBroker) ExchangeDeclare(exchangeName string, kind string, opts ...ExchangeDeclareOption) cerror.AppError
  • func (l *RabbitBroker) Publish(exchange string, routingKey string, body []byte, userId string, t time.Time) cerror.AppError
  • func (l *RabbitBroker) QueueBind(exchangeName string, queueName string, routingKey string, noWait bool, ...) cerror.AppError
  • func (l *RabbitBroker) QueueDeclare(queueName string, opts ...QueueDeclareOption) (string, cerror.AppError)
  • func (l *RabbitBroker) QueueStartConsume(queueName string, consumerName string, handleFunc HandleFunc, ...) cerror.AppError
  • func (l *RabbitBroker) QueueStopConsume(queueName string) cerror.AppError
  • func (l *RabbitBroker) Start() cerror.AppError
  • func (l *RabbitBroker) StartAllConsumers() cerror.AppError
  • func (l *RabbitBroker) Stop()
  • func (l *RabbitBroker) StopAllConsumers() cerror.AppError

Constants

const MaxReconnectAttempts = 100

Functions

func BuildRabbitQueueConsumer

func BuildRabbitQueueConsumer(delivery <-chan amqp.Delivery, handleFunc HandleFunc, consumerName string, handleTimeout time.Duration) (*rabbitQueueConsumer, cerror.AppError)

func ExchangeEnableAutoDelete

func ExchangeEnableAutoDelete(o *ExchangeDeclareOptions)

func ExchangeEnableDurable

func ExchangeEnableDurable(o *ExchangeDeclareOptions)

func ExchangeEnableInternal

func ExchangeEnableInternal(o *ExchangeDeclareOptions)

func ExchangeEnableNoWait

func ExchangeEnableNoWait(o *ExchangeDeclareOptions)

func QueueEnableAutoDelete

func QueueEnableAutoDelete(o *QueueDeclareOptions)

func QueueEnableDurable

func QueueEnableDurable(o *QueueDeclareOptions)

func QueueEnableExclusive

func QueueEnableExclusive(o *QueueDeclareOptions)

func QueueEnableNoWait

func QueueEnableNoWait(o *QueueDeclareOptions)

Types

type ExchangeDeclareOption

type ExchangeDeclareOption func(*ExchangeDeclareOptions)

type ExchangeDeclareOptions

type ExchangeDeclareOptions struct {
	Args       map[string]any
	Durable    bool
	AutoDelete bool
	Internal   bool
	NoWait     bool
}

type HandleFunc

type HandleFunc func(handleTimeout time.Duration, delivery <-chan amqp.Delivery, stopper chan any)

AcknowledgeFunc allows to define the reaction to the amqp.Delivery.

Will run in goroutine and should handle logic for the acknowledging messages.

delivery - channel where amqp.messages will be delivered

stopper - channel for stopping the routine

handleFunc - function used to handle the exact amqp.message content

type QueueDeclareOption

type QueueDeclareOption func(options *QueueDeclareOptions)

type QueueDeclareOptions

type QueueDeclareOptions struct {
	Args       map[string]any
	Durable    bool
	AutoDelete bool
	Exclusive  bool
	NoWait     bool
}

type RabbitBroker

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

func BuildRabbit

func BuildRabbit(config *conf.RabbitConfig, errChan chan<- cerror.AppError) (*RabbitBroker, cerror.AppError)

func (*RabbitBroker) Consume

func (l *RabbitBroker) Consume(queueName string, consumerName string) (<-chan amqp.Delivery, cerror.AppError)

func (*RabbitBroker) ExchangeBind

func (l *RabbitBroker) ExchangeBind(destination string, key string, source string, noWait bool, args map[string]any) cerror.AppError

func (*RabbitBroker) ExchangeDeclare

func (l *RabbitBroker) ExchangeDeclare(exchangeName string, kind string, opts ...ExchangeDeclareOption) cerror.AppError

func (*RabbitBroker) Publish

func (l *RabbitBroker) Publish(
	exchange string,
	routingKey string,
	body []byte,
	userId string,
	t time.Time,
) cerror.AppError

func (*RabbitBroker) QueueBind

func (l *RabbitBroker) QueueBind(exchangeName string, queueName string, routingKey string, noWait bool, args map[string]any) cerror.AppError

func (*RabbitBroker) QueueDeclare

func (l *RabbitBroker) QueueDeclare(queueName string, opts ...QueueDeclareOption) (string, cerror.AppError)

func (*RabbitBroker) QueueStartConsume

func (l *RabbitBroker) QueueStartConsume(queueName string, consumerName string, handleFunc HandleFunc, handleTimeout time.Duration) cerror.AppError

func (*RabbitBroker) QueueStopConsume

func (l *RabbitBroker) QueueStopConsume(queueName string) cerror.AppError

func (*RabbitBroker) Start

func (l *RabbitBroker) Start() cerror.AppError

Start starts the channel between rabbitMQ server and this server

func (*RabbitBroker) StartAllConsumers

func (l *RabbitBroker) StartAllConsumers() cerror.AppError

StartAllConsumers starts all consumers if exist

func (*RabbitBroker) Stop

func (l *RabbitBroker) Stop()

Stop stops all consumers and connections of rabbit

func (*RabbitBroker) StopAllConsumers

func (l *RabbitBroker) StopAllConsumers() cerror.AppError

StopAllConsumers stops all consumers if exist