rabbitmq

Documentation

Index

• Constants
• func SetupTrackerQueue(queueName string)
• type AMQPConfig
  • func (me *AMQPConfig) Clone() AMQPConfig
  • func (me *AMQPConfig) Equals(to *AMQPConfig) bool
  • func (me *AMQPConfig) IsBroadcastExange() bool
• type AMQPConsumerCallback
• type RabbitConsumerProxy
  • func GenerateRabbitMQConsumerProxy(consumeProxy *mqenv.MQConsumerProxy, exchangeName string) *RabbitConsumerProxy
• type RabbitMQ
  • func GetRabbitMQ(name string) (*RabbitMQ, error)
  • func InitRabbitMQ(mqConnName string, connCfg *mqenv.MQConnectorConfig, amqpCfg *AMQPConfig) (*RabbitMQ, error)
  • func NewRabbitMQ(mqConnName string, connCfg *mqenv.MQConnectorConfig, amqpCfg *AMQPConfig) *RabbitMQ
  • func (r *RabbitMQ) CheckQueueConsumers(queueName string) int
  • func (r *RabbitMQ) QueryRPC(pm *mqenv.MQPublishMessage) (*mqenv.MQConsumerMessage, error)
  • func (r *RabbitMQ) Run()
• type RabbitQueueStatus
• type RabbitRPC
  • func GetRPCRabbitMQWithConsumers(key string) *RabbitRPC
  • func GetRPCRabbitMQWithoutConnectedChecking(key string) *RabbitRPC
  • func InitRPCRabbitMQ(key string, rpcType int, connCfg *mqenv.MQConnectorConfig, amqpCfg *AMQPConfig) *RabbitRPC

Constants

const (
	// AMQPReconnectDuration reconnect duration
	AMQPReconnectDuration = 1
	// AMQPQueueStatusFreshDuration queue status refresh duration
	AMQPQueueStatusFreshDuration = 30
)

const (
	ChannelConsumerDeliveryCheckIntervalSeconds = int64(9)
	JinDie                                      = "_jindie_mq" // 获取金蝶环境变量的名字

)

Constants

Functions

func SetupTrackerQueue

func SetupTrackerQueue(queueName string)

SetupTrackerQueue name

Types

type AMQPConfig

type AMQPConfig struct {
	ConnConfigName  string
	Queue           string
	QueueDurable    bool
	BindingExchange bool
	ExchangeName    string
	ExchangeType    string
	BindingKey      string
	QueueAutoDelete bool
}

AMQPConfig queue config

func (*AMQPConfig) Clone

func (me *AMQPConfig) Clone() AMQPConfig

Clone a AMQPConfig object

func (*AMQPConfig) Equals

func (me *AMQPConfig) Equals(to *AMQPConfig) bool

Equals check if equals

func (*AMQPConfig) IsBroadcastExange

func (me *AMQPConfig) IsBroadcastExange() bool

IsBroadcastExange check if the configure is fanout

type AMQPConsumerCallback

type AMQPConsumerCallback func(amqp.Delivery) *mqenv.MQPublishMessage

AMQPConsumerCallback callback

type RabbitConsumerProxy

type RabbitConsumerProxy struct {
	Queue       string
	Callback    AMQPConsumerCallback
	ConsumerTag string
	AutoAck     bool
	Exclusive   bool
	NoLocal     bool
	NoWait      bool
	Arguments   amqp.Table
	// contains filtered or unexported fields
}

RabbitConsumerProxy consumer proxy

func GenerateRabbitMQConsumerProxy

func GenerateRabbitMQConsumerProxy(consumeProxy *mqenv.MQConsumerProxy, exchangeName string) *RabbitConsumerProxy

GenerateRabbitMQConsumerProxy generate rabbitmq consumer proxy

type RabbitMQ

type RabbitMQ struct {
	Name       string
	Publish    chan *mqenv.MQPublishMessage
	Consume    chan *RabbitConsumerProxy
	Done       chan error
	Channel    *amqp.Channel
	Conn       *amqp.Connection
	Config     *AMQPConfig
	ConnConfig *mqenv.MQConnectorConfig
	Close      chan interface{}
	// contains filtered or unexported fields
}

RabbitMQ instance

func GetRabbitMQ

func GetRabbitMQ(name string) (*RabbitMQ, error)

GetRabbitMQ get

func InitRabbitMQ

func InitRabbitMQ(mqConnName string, connCfg *mqenv.MQConnectorConfig, amqpCfg *AMQPConfig) (*RabbitMQ, error)

InitRabbitMQ init

func NewRabbitMQ

func NewRabbitMQ(mqConnName string, connCfg *mqenv.MQConnectorConfig, amqpCfg *AMQPConfig) *RabbitMQ

NewRabbitMQ with parameters

func (*RabbitMQ) CheckQueueConsumers

func (r *RabbitMQ) CheckQueueConsumers(queueName string) int

CheckQueueConsumers check if queue has consumers listening

func (*RabbitMQ) QueryRPC

func (r *RabbitMQ) QueryRPC(pm *mqenv.MQPublishMessage) (*mqenv.MQConsumerMessage, error)

QueryRPC publishes a message and waiting the response

func (*RabbitMQ) Run

func (r *RabbitMQ) Run()

Run start 1. init the rabbitmq conneciton 2. expect messages from the message hub on the Publish channel 3. if the connection is closed, try to restart it

type RabbitQueueStatus

type RabbitQueueStatus struct {
	RefreshingTime int64
	QueueName      string
	Consumers      int
	Messages       int
}

RabbitQueueStatus queue status

type RabbitRPC

type RabbitRPC struct {
	RabbitMQ
	Deliveries <-chan amqp.Delivery
	RPCType    int
}

RabbitRPC rpc instance

func GetRPCRabbitMQWithConsumers

func GetRPCRabbitMQWithConsumers(key string) *RabbitRPC

GetRPCRabbitMQWithConsumers get instance

func GetRPCRabbitMQWithoutConnectedChecking

func GetRPCRabbitMQWithoutConnectedChecking(key string) *RabbitRPC

GetRPCRabbitMQWithoutConnectedChecking get instance

func InitRPCRabbitMQ

func InitRPCRabbitMQ(key string, rpcType int, connCfg *mqenv.MQConnectorConfig, amqpCfg *AMQPConfig) *RabbitRPC

InitRPCRabbitMQ init