rq

Documentation

Index

• Constants
• func NewRabbitmq(rabbitmqConf RabbitmqConf) error
• func VerifyClient() error
• type Client
  • func NewClient(isPermanent ...bool) Client
• type Consume
  • func NewConsume(queueName string, ...) Consume
• type Pusher
  • func NewPusher() Pusher
• type Rabbitmq
• type RabbitmqConf

Constants

const (
	DeadLetterExchange   = "x-dead-letter-exchange"
	DeadLetterRoutingKey = "x-dead-letter-routing-key"
)

Functions

func NewRabbitmq

func NewRabbitmq(rabbitmqConf RabbitmqConf) error

func VerifyClient

func VerifyClient() error

Types

type Client

type Client interface {
	Connect(isPermanent bool) error
	CloseConnection()
	HandleNotify()
	OpenChannel() (*amqp.Channel, error)
	CloseChannel(ch *amqp.Channel)
	Push(queueName string, data *qtypes.MessageData) error
	Consume(channel *amqp.Channel, queueName string) (<-chan amqp.Delivery, error)
	Close()
}

func NewClient

func NewClient(isPermanent ...bool) Client

type Consume

type Consume interface {
	// Start consume
	Start()
	// Close consume
	Close()
}

func NewConsume

func NewConsume(queueName string, handler func(queueName string, data *qtypes.MessageData) error) Consume

NewConsume new consume

type Pusher

type Pusher interface {
	// Send queue message
	// data queue message
	Send(data *qtypes.QueueDelayData) error
	// Close pusher
	Close()
}

func NewPusher

func NewPusher() Pusher

type Rabbitmq

type Rabbitmq interface {
	GetExchange() string
	GetConsumerTimeout() int64
	GetAmqpURI() string
	GetQueueMap() map[string]bool
	GetMaxRetry() uint16
	IsSingleActive(queueName string) bool
	Verify() error
	Initialize()
}

type RabbitmqConf

type RabbitmqConf struct {
	Enable bool `json:",optional"`
	// max retry num
	MaxRetries uint16 `json:",optional"`
	// server address ip:port
	Broker   string `json:",optional"`
	Username string `json:",optional"`
	Password string `json:",optional"`
	Vhost    string `json:",optional"`
	Exchange string `json:",optional"`
	NonBlock bool   `json:",optional,default=true"`
	// queue config
	Queue []qtypes.QueueConf `json:",optional"`
	// Delivery Acknowledgement Timeout(unit millisecond)
	ConsumerTimeout int64 `json:",optional"`
}