rabbitmq

Documentation

Index

• Constants
• type AMQPConfig
  • func (me *AMQPConfig) Equals(to *AMQPConfig) bool
  • func (me *AMQPConfig) IsBroadcaseExange() bool
• type AMQPConsumerCallback
• type RabbitConsumerProxy
  • func GenerateRabbitMQConsumerProxy(consumeProxy *mqenv.MQConsumerProxy) *RabbitConsumerProxy
• type RabbitMQ
  • func GetRabbitMQ(name string) (*RabbitMQ, error)
  • func InitRabbitMQ(mqConnName string, connCfg *mqenv.MQConnectorConfig, amqpCfg *AMQPConfig) (*RabbitMQ, error)
  • func (r *RabbitMQ) Run()
• type RabbitPublishingMsg
  • func GenerateRabbitMQPublishMessage(publishMsg *mqenv.MQPublishMessage) *RabbitPublishingMsg
• type RabbitQueueStatus
• type RabbitRPCMQ
  • func GetRPCRabbitMQ(key string) *RabbitRPCMQ
  • func InitRPCRabbitMQ(key string, rpcType int, connCfg *mqenv.MQConnectorConfig, amqpCfg *AMQPConfig) *RabbitRPCMQ
  • func (r *RabbitRPCMQ) Run()
• type RabbitRPCMsgProxy
  • func (m *RabbitRPCMsgProxy) OnClosed()

Constants

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

Types

type AMQPConfig

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

AMQPConfig queue config

func (*AMQPConfig) Equals

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

Equals check if equals

func (*AMQPConfig) IsBroadcaseExange

func (me *AMQPConfig) IsBroadcaseExange() bool

IsBroadcaseExange check if the configure is fanout

type AMQPConsumerCallback

type AMQPConsumerCallback func(amqp.Delivery)

AMQPConsumerCallback callback

type RabbitConsumerProxy

type RabbitConsumerProxy struct {
	Queue       string
	Callback    AMQPConsumerCallback
	ConsumerTag string
	AutoAck     bool
	Exclusive   bool
	NoLocal     bool
	NoWait      bool
	Arguments   amqp.Table
}

RabbitConsumerProxy consumer proxy

func GenerateRabbitMQConsumerProxy

func GenerateRabbitMQConsumerProxy(consumeProxy *mqenv.MQConsumerProxy) *RabbitConsumerProxy

GenerateRabbitMQConsumerProxy generate rabbitmq consumer proxy

type RabbitMQ

type RabbitMQ struct {
	Name       string
	Publish    chan *RabbitPublishingMsg
	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 (*RabbitMQ) Run

func (r *RabbitMQ) Run()

Run start

type RabbitPublishingMsg

type RabbitPublishingMsg struct {
	Body          []byte
	RoutingKey    string
	CorrelationID string             `json:"correlationId"`
	ReplyTo       string             `json:"replyTo"`
	PublishStatus chan mqenv.MQEvent `json:"-"`
	EventLabel    string             `json:"eventLabel"`
	Headers       map[string]string  `json:"headers"`
}

RabbitPublishingMsg publishing message

func GenerateRabbitMQPublishMessage

func GenerateRabbitMQPublishMessage(publishMsg *mqenv.MQPublishMessage) *RabbitPublishingMsg

GenerateRabbitMQPublishMessage generate publish message

type RabbitQueueStatus

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

RabbitQueueStatus queue status

type RabbitRPCMQ

type RabbitRPCMQ struct {
	Name        string
	Publish     chan *RabbitRPCMsgProxy
	Deliveries  <-chan amqp.Delivery
	Done        chan error
	Channel     *amqp.Channel
	Conn        *amqp.Connection
	QueueStatus *RabbitQueueStatus
	Config      *AMQPConfig
	ConnConfig  *mqenv.MQConnectorConfig
	Close       chan interface{}
	RPCType     int
	// contains filtered or unexported fields
}

RabbitRPCMQ rpc instance

func GetRPCRabbitMQ

func GetRPCRabbitMQ(key string) *RabbitRPCMQ

GetRPCRabbitMQ get instance

func InitRPCRabbitMQ

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

InitRPCRabbitMQ init

func (*RabbitRPCMQ) Run

func (r *RabbitRPCMQ) 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 RabbitRPCMsgProxy

type RabbitRPCMsgProxy struct {
	Request      string
	Response     chan string
	ReplyToQueue string
	// contains filtered or unexported fields
}

RabbitRPCMsgProxy rpc proxy

func (*RabbitRPCMsgProxy) OnClosed

func (m *RabbitRPCMsgProxy) OnClosed()

OnClosed on close event