demo/rabbitmq/RabbitMQ/entity.go

package RabbitMQ

import "github.com/streadway/amqp"

// rabbitMQ结构体
type RabbitMQ struct {
	Conn             *amqp.Connection
	ConsumerChannel  *amqp.Channel
	PublisherChannel *amqp.Channel
	QueueConfig      QueueConfig
	ExchangeConfig   ExchangeConfig
	Consumer
	Publisher
}

type Opt func()

type Consumer struct {
	autoAck   bool       //是否自动应答
	exclusive bool       //是否独有
	noLocal   bool       // //设置为true，表示 不能将同一个Conenction中生产者发送的消息传递给这个Connection中 的消费者
	noWait    bool       //列是否阻塞
	args      amqp.Table // 额外的属性
}

// AutoAck 自动应答
func (c *Consumer) AutoAck() Opt {
	return func() {
		c.autoAck = true
	}
}

// Exclusive 独有
func (c *Consumer) Exclusive() Opt {
	return func() {
		c.exclusive = true
	}
}

// NoLocal 设置为true，表示不能将同一个Conenction中生产者发送的消息传递给这个Connection中的消费者
func (c *Consumer) NoLocal() Opt {
	return func() {
		c.noLocal = true
	}
}

// NoWait 是否阻塞
func (c *Consumer) NoWait() Opt {
	return func() {
		c.noWait = true
	}
}

type Publisher struct {
	mandatory bool //
	immediate bool //
}

// Mandatory  是否将无法路由到任何队列的消息返回
func (p *Publisher) Mandatory() Opt {
	return func() {
		p.mandatory = true
	}
}

// Immediate 当exchange发送消息到队列后发现队列上没有消费者，则会把消息返还给发送者
func (p *Publisher) Immediate() Opt {
	return func() {
		p.immediate = true
	}
}

// 配置
type QueueConfig struct {
	durable    bool       // 是否持久化
	autoDelete bool       // 是否自动删除
	exclusive  bool       // 是否独占连接
	noWait     bool       // 是否阻塞处理
	args       amqp.Table // 额外的属性
}

// Durable 队列持久化
func (c *QueueConfig) Durable() Opt {
	return func() {
		c.durable = true
	}
}

// AutoDelete 队列自动删除
func (c *QueueConfig) AutoDelete() Opt {
	return func() {
		c.autoDelete = true
	}
}

// Exclusive 队列独占连接
func (c *QueueConfig) Exclusive() Opt {
	return func() {
		c.exclusive = true
	}
}

// NoWait 是否阻塞
func (c *QueueConfig) NoWait() Opt {
	return func() {
		c.noWait = true
	}
}

// Args 额外的属性
func (c *QueueConfig) Args(args amqp.Table) Opt {
	return func() {
		c.args = args
	}
}

type ExchangeConfig struct {
	durable    bool       //是否持久化
	autoDelete bool       //是否自动删除
	internal   bool       // 是否内部处理
	noWait     bool       //是否阻塞处理
	args       amqp.Table //额外的属性
}

// Durable 交换机持久化
func (c ExchangeConfig) Durable() Opt {
	return func() {
		c.durable = true
	}
}

// AutoDelete 交换机自动删除
func (c *ExchangeConfig) AutoDelete() Opt {
	return func() {
		c.autoDelete = true
	}
}

// Internal 是否内部处理
func (c *ExchangeConfig) Internal() Opt {
	return func() {
		c.internal = true
	}
}

// NoWait 是否阻塞
func (c *ExchangeConfig) NoWait() Opt {
	return func() {
		c.noWait = true
	}
}

// Args 额外的属性
func (c *ExchangeConfig) Args(args amqp.Table) Opt {
	return func() {
		c.args = args
	}
}