rabbitmq

Documentation

Index

• Constants
• type Config
• type Consumer
  • func (c *Consumer) CloseAll() (err error)
  • func (c *Consumer) Consume()
  • func (c *Consumer) ConsumingMessage() (chan amqp.Delivery, chan error)
  • func (c *Consumer) Reconnect() (err error)
• type IConsumer
  • func NewConsumer(opts ...Option) IConsumer
• type IPublisher
  • func NewPublisher(opts ...Option) IPublisher
• type IRabbitMQ
  • func New(conf *Config) (IRabbitMQ, error)
• type Option
  • func WithConfig(config *Config) Option
  • func WithErrorHandleFn(errHandleFn func(err error)) Option
  • func WithHandleFn(handleFn func(msg amqp.Delivery) error) Option
  • func WithThreads(threads int) Option
• type Publisher
  • func (pub *Publisher) Publish(payload []byte, routingKey string, reliable bool) (err error)
• type RabbitMQ
  • func (mq *RabbitMQ) BindQueue(exchange, routingKey string) error
  • func (mq *RabbitMQ) ChanelIsClosed() bool
  • func (mq *RabbitMQ) CloseChannel() error
  • func (mq *RabbitMQ) CloseConnection() error
  • func (mq *RabbitMQ) DeclareExchange() error
  • func (mq *RabbitMQ) DeclareQueue() error
  • func (mq *RabbitMQ) EnsureConnection() (err error)
  • func (mq *RabbitMQ) NewChannel() (*amqp.Channel, error)
  • func (mq *RabbitMQ) NewConnection() (*amqp.Connection, error)
  • func (mq *RabbitMQ) Setup()

Constants

const (
	ConsumerThreads = 10
)

const (
	WaitTimeReconnect = 5
)

Rabbitmq constant

Types

type Config

type Config struct {
	AMQPUrl      string `json:"amqp_url"`
	Host         string `json:"host"`
	Port         string `json:"port"`
	Vhost        string `json:"vhost"`
	Username     string `json:"username"`
	Password     string `json:"password"`
	ExchangeName string `json:"exchange_name"`
	ExchangeType string `json:"exchange_type"`
	QueueName    string `json:"queue_name"`
}

type Consumer

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

func (*Consumer) CloseAll

func (c *Consumer) CloseAll() (err error)

CloseAll : Close connection and channel

func (*Consumer) Consume

func (c *Consumer) Consume()

func (*Consumer) ConsumingMessage

func (c *Consumer) ConsumingMessage() (chan amqp.Delivery, chan error)

func (*Consumer) Reconnect

func (c *Consumer) Reconnect() (err error)

type IConsumer

type IConsumer interface {
	Consume()
}

func NewConsumer

func NewConsumer(opts ...Option) IConsumer

type IPublisher

type IPublisher interface {
	Publish(payload []byte, routingKey string, reliable bool) (err error)
}

func NewPublisher

func NewPublisher(opts ...Option) IPublisher

type IRabbitMQ

type IRabbitMQ interface {
	NewConnection() (*amqp.Connection, error)
	CloseConnection() error
	NewChannel() (*amqp.Channel, error)
	EnsureConnection() (err error)
	CloseChannel() error
	DeclareExchange() error
	DeclareQueue() error
	BindQueue(exchange, routingKey string) error
	Setup()
	ChanelIsClosed() bool
}

func New

func New(conf *Config) (IRabbitMQ, error)

type Option

type Option interface {
	// contains filtered or unexported methods
}

Option rabbitmq option

func WithConfig

func WithConfig(config *Config) Option

WithConfig set config

func WithErrorHandleFn

func WithErrorHandleFn(errHandleFn func(err error)) Option

WithErrorHandleFn set errHandleFn

func WithHandleFn

func WithHandleFn(handleFn func(msg amqp.Delivery) error) Option

WithHandleFn set handleFn

func WithThreads

func WithThreads(threads int) Option

WithThreads set threads

type Publisher

type Publisher struct {
	*RabbitMQ
}

func (*Publisher) Publish

func (pub *Publisher) Publish(payload []byte, routingKey string, reliable bool) (err error)

type RabbitMQ

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

func (*RabbitMQ) BindQueue

func (mq *RabbitMQ) BindQueue(exchange, routingKey string) error

func (*RabbitMQ) ChanelIsClosed

func (mq *RabbitMQ) ChanelIsClosed() bool

func (*RabbitMQ) CloseChannel

func (mq *RabbitMQ) CloseChannel() error

func (*RabbitMQ) CloseConnection

func (mq *RabbitMQ) CloseConnection() error

func (*RabbitMQ) DeclareExchange

func (mq *RabbitMQ) DeclareExchange() error

func (*RabbitMQ) DeclareQueue

func (mq *RabbitMQ) DeclareQueue() error

func (*RabbitMQ) EnsureConnection

func (mq *RabbitMQ) EnsureConnection() (err error)

func (*RabbitMQ) NewChannel

func (mq *RabbitMQ) NewChannel() (*amqp.Channel, error)

func (*RabbitMQ) NewConnection

func (mq *RabbitMQ) NewConnection() (*amqp.Connection, error)

func (*RabbitMQ) Setup

func (mq *RabbitMQ) Setup()