package rabbitmq
import (
	"encoding/json"
	"github.com/streadway/amqp"
	"log"
)
// RabbitMQ RabbitMQ結(jié)構(gòu)
type RabbitMQ struct {
	channel  *amqp.Channel
	Name     string
	exchange string
}
// Connect 連接服務(wù)器
func Connect(s string) *RabbitMQ {
	//連接rabbitmq
	conn, e := amqp.Dial(s)
	failOnError(e, "連接Rabbitmq服務(wù)器失??！")
	ch, e := conn.Channel()
	failOnError(e, "無(wú)法打開(kāi)頻道！")
	mq := new(RabbitMQ)
	mq.channel = ch
	return mq
}
// New 初始化消息隊(duì)列
//第一個(gè)參數(shù)：rabbitmq服務(wù)器的鏈接，第二個(gè)參數(shù)：隊(duì)列名字
func New(s string, name string) *RabbitMQ {
	//連接rabbitmq
	conn, e := amqp.Dial(s)
	failOnError(e, "連接Rabbitmq服務(wù)器失??！")
	ch, e := conn.Channel()
	failOnError(e, "無(wú)法打開(kāi)頻道！")
	q, e := ch.QueueDeclare(
		name,  //隊(duì)列名
		false, //是否開(kāi)啟持久化
		true,  //不使用時(shí)刪除
		false, //排他
		false, //不等待
		nil,   //參數(shù)
	)
	failOnError(e, "初始化消息隊(duì)列失??！")
	mq := new(RabbitMQ)
	mq.channel = ch
	mq.Name = q.Name
	return mq
}
// QueueDeclare 聲明queue
func (q *RabbitMQ) QueueDeclare(queue string) {
	_, e := q.channel.QueueDeclare(queue, false, true, false, false, nil)
	failOnError(e, "聲明queue失敗！")
}
// QueueDelete 刪除queue
func (q *RabbitMQ) QueueDelete(queue string) {
	_, e := q.channel.QueueDelete(queue, false, true, false)
	failOnError(e, "刪除queue失??！")
}
// Qos 配置queue參數(shù)
func (q *RabbitMQ) Qos() {
	e := q.channel.Qos(1, 0, false)
	failOnError(e, "無(wú)法設(shè)置QoS")
}
// NewExchange 初始化交換機(jī)
//第一個(gè)參數(shù)：rabbitmq服務(wù)器的鏈接，第二個(gè)參數(shù)：交換機(jī)名字，第三個(gè)參數(shù)：交換機(jī)類(lèi)型
func NewExchange(s string, name string, typename string) {
	//連接rabbitmq
	conn, e := amqp.Dial(s)
	failOnError(e, "連接Rabbitmq服務(wù)器失??！")
	ch, e := conn.Channel()
	failOnError(e, "無(wú)法打開(kāi)頻道！")
	e = ch.ExchangeDeclare(
		name,     // name
		typename, // type
		true,     // durable
		false,    // auto-deleted
		false,    // internal
		false,    // no-wait
		nil,      // arguments
	)
	failOnError(e, "初始化交換機(jī)失?。?)
}
// ExchangeDelete 刪除交換機(jī)
func (q *RabbitMQ) ExchangeDelete(exchange string) {
	e := q.channel.ExchangeDelete(exchange, false, true)
	failOnError(e, "刪除交換機(jī)失??！")
}
// Bind 綁定消息隊(duì)列到exchange
func (q *RabbitMQ) Bind(exchange string, key string) {
	e := q.channel.QueueBind(
		q.Name,
		key,
		exchange,
		false,
		nil,
	)
	failOnError(e, "綁定隊(duì)列失??！")
	q.exchange = exchange
}
// Send 向消息隊(duì)列發(fā)送消息
//Send方法可以往某個(gè)消息隊(duì)列發(fā)送消息
func (q *RabbitMQ) Send(queue string, body interface{}) {
	str, e := json.Marshal(body)
	failOnError(e, "消息序列化失?。?)
	e = q.channel.Publish(
		"",    //交換
		queue, //路由鍵
		false, //必填
		false, //立即
		amqp.Publishing{
			ReplyTo: q.Name,
			Body:    []byte(str),
		})
	msg := "向隊(duì)列:" + q.Name + "發(fā)送消息失??！"
	failOnError(e, msg)
}
// Publish 向exchange發(fā)送消息
//Publish方法可以往某個(gè)exchange發(fā)送消息
func (q *RabbitMQ) Publish(exchange string, body interface{}, key string) {
	str, e := json.Marshal(body)
	failOnError(e, "消息序列化失??！")
	e = q.channel.Publish(
		exchange,
		key,
		false,
		false,
		amqp.Publishing{ReplyTo: q.Name,
			Body: []byte(str)},
	)
	failOnError(e, "向交換機(jī)發(fā)送消息失敗！")
}
// Consume 接收某個(gè)消息隊(duì)列的消息
func (q *RabbitMQ) Consume() <-chan amqp.Delivery {
	c, e := q.channel.Consume(
		q.Name, //指定從哪個(gè)隊(duì)列中接收消息
		"",
		true,
		false,
		false,
		false,
		nil,
	)
	failOnError(e, "接收消息失?。?)
	return c
}
// Close 關(guān)閉隊(duì)列連接
func (q *RabbitMQ) Close() {
	q.channel.Close()
}
//錯(cuò)誤處理函數(shù)
func failOnError(err error, msg string) {
	if err != nil {
		log.Fatalf("%s: %s", msg, err)
	}
}

func main() {
	//第一個(gè)參數(shù)指定rabbitmq服務(wù)器的鏈接，第二個(gè)參數(shù)指定創(chuàng)建隊(duì)列的名字
	producer := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "simple")
	i := 0
	for {
		// 每隔2s發(fā)送一次消息
		time.Sleep(time.Second * 2)
		producer.Send("simple", " simple message: "+strconv.Itoa(i))
		i = i + 1
	}
}

func main() {
	consumer := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "simple")
	//接收消息時(shí)，指定
	messages := consumer.Consume()
	go func() {
		for ch := range messages {
			log.Printf("Received a message: %s", ch.Body)
			// 消費(fèi)消息要用3s
			time.Sleep(time.Second * 3)
		}
	}()
	select {}
}

func main() {
	//第一個(gè)參數(shù)指定rabbitmq服務(wù)器的鏈接，第二個(gè)參數(shù)指定創(chuàng)建隊(duì)列的名字
	producer := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "worker")
	i := 0
	for {
		// 每隔2s發(fā)送一次消息
		time.Sleep(time.Second * 2)
		producer.Send("worker", " worker message: "+strconv.Itoa(i))
		i = i + 1
	}
}

func main() {
	consumer1 := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "worker")
	//接收消息
	messages := consumer1.Consume()
	go func() {
		for ch := range messages {
			log.Printf("Received a message: %s", ch.Body)
			// 消費(fèi)消息要用3s
			time.Sleep(time.Second * 3)
		}
	}()
	select {}
}

func main() {
	consumer2 := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "worker")
	//接收消息
	messages := consumer2.Consume()
	go func() {
		for ch := range messages {
			log.Printf("Received a message: %s", ch.Body)
			// 消費(fèi)消息要用3s
			time.Sleep(time.Second * 3)
		}
	}()
	select {}
}

// Consume 接收某個(gè)消息隊(duì)列的消息
func (q *RabbitMQ) Consume() <-chan amqp.Delivery {
	c, e := q.channel.Consume(
		q.Name, //指定從哪個(gè)隊(duì)列中接收消息
		"",
		false, // 不自動(dòng)確認(rèn)消息
		false,
		false,
		false,
		nil,
	)
	failOnError(e, "接收消息失??！")
	return c
}

func main() {
	//第一個(gè)參數(shù)指定rabbitmq服務(wù)器的鏈接，第二個(gè)參數(shù)指定創(chuàng)建隊(duì)列的名字
	producer := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "worker")
	i := 0
	for {
		// 每隔2s發(fā)送一次消息
		time.Sleep(time.Second * 2)
		producer.Send("worker", " worker message: "+strconv.Itoa(i))
		i = i + 1
	}
}

// Qos 配置queue參數(shù)
func (q *RabbitMQ) Qos() {
	e := q.channel.Qos(1, 0, false)
	failOnError(e, "無(wú)法設(shè)置QoS")
}

func main() {
	consumer1 := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "worker")
	// 指定一次只消費(fèi)一條消息，直到消費(fèi)完才重新接收
	consumer1.Qos()
	//接收消息
	messages := consumer1.Consume()
	go func() {
		for ch := range messages {
			log.Printf("Received a message: %s", ch.Body)
			// 消費(fèi)消息要用10s
			time.Sleep(time.Second * 10)
			// 手動(dòng)回復(fù)
			ch.Ack(false)
		}
	}()
	select {}
}

func main() {
	consumer2 := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "worker")
	// 指定一次只消費(fèi)一條消息，直到消費(fèi)完才重新接收
	consumer2.Qos()
	//接收消息
	messages := consumer2.Consume()
	go func() {
		for ch := range messages {
			log.Printf("Received a message: %s", ch.Body)
			// 消費(fèi)消息要用2s
			time.Sleep(time.Second * 2)
			// 手動(dòng)回復(fù)
			ch.Ack(false)
		}
	}()
	select {}
}

func main() {
	producer := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "queue")
	rabbitmq.NewExchange("amqp://guest:guest@35.76.111.125:5672/", "exchange1", "fanout")
	i := 0
	for {
		time.Sleep(time.Second)
		// fanout模式下不用routing key
		producer.Publish("exchange1", "pubsub message: "+strconv.Itoa(i), "")
		i = i + 1
	}
}

func main() {
	//第一個(gè)參數(shù)指定rabbitmq服務(wù)器的鏈接，第二個(gè)參數(shù)指定創(chuàng)建隊(duì)列的名字
	consumer1 := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "queue1")
	// 隊(duì)列綁定到exchange
	consumer1.Bind("exchange1", "")
	//接收消息
	msgs := consumer1.Consume()
	go func() {
		for d := range msgs {
			log.Printf("Consumer1 received a message: %s", d.Body)
			d.Ack(false)
		}
	}()
	select {}
}

func main() {
	//第一個(gè)參數(shù)指定rabbitmq服務(wù)器的鏈接，第二個(gè)參數(shù)指定創(chuàng)建隊(duì)列的名字
	consumer2 := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "queue2")
	// 隊(duì)列綁定到exchange
	consumer2.Bind("exchange1", "")
	//接收消息
	msgs := consumer2.Consume()
	go func() {
		for d := range msgs {
			log.Printf("Consumer2 received a message: %s", d.Body)
			d.Ack(false)
		}
	}()
	select {}
}

func main() {
	producer := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "queue")
	// 指定為direct類(lèi)型
	rabbitmq.NewExchange("amqp://guest:guest@35.76.111.125:5672/", "exchange", "direct")
	i := 0
	for {
		time.Sleep(time.Second)
		// 如果是奇數(shù)，就發(fā)key1
		// 如果是偶數(shù)，就發(fā)key2
		if i%2 != 0 {
			producer.Publish("exchange", "routing message: "+strconv.Itoa(i), "key1")
		} else {
			producer.Publish("exchange", "routing message: "+strconv.Itoa(i), "key2")
		}
		i = i + 1
	}
}

func main() {
	//第一個(gè)參數(shù)指定rabbitmq服務(wù)器的鏈接，第二個(gè)參數(shù)指定創(chuàng)建隊(duì)列的名字
	consumer1 := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "queue1")
	// 隊(duì)列綁定到exchange
	consumer1.Bind("exchange", "key1")
	//接收消息
	msgs := consumer1.Consume()
	go func() {
		for d := range msgs {
			log.Printf("Consumer1 received a message: %s", d.Body)
			d.Ack(false)
		}
	}()
	select {}
}

func main() {
	//第一個(gè)參數(shù)指定rabbitmq服務(wù)器的鏈接，第二個(gè)參數(shù)指定創(chuàng)建隊(duì)列的名字
	consumer2 := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "queue2")
	// 隊(duì)列綁定到exchange
	consumer2.Bind("exchange", "key2")
	//接收消息
	msgs := consumer2.Consume()
	go func() {
		for d := range msgs {
			log.Printf("Consumer2 received a message: %s", d.Body)
			d.Ack(false)
		}
	}()
	select {}
}

func main() {
	producer := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "queue")
	// 指定為topic類(lèi)型
	rabbitmq.NewExchange("amqp://guest:guest@35.76.111.125:5672/", "exchange2", "topic")
	var i int
	for {
		time.Sleep(time.Second)
		if i%2 != 0 {
			producer.Publish("exchange2", "topic message: "+strconv.Itoa(i), "a.test.b.c")
		} else {
			producer.Publish("exchange2", "topic message: "+strconv.Itoa(i), "a.test.b")
		}
		i++
	}
}

func main() {
	//第一個(gè)參數(shù)指定rabbitmq服務(wù)器的鏈接，第二個(gè)參數(shù)指定創(chuàng)建隊(duì)列的名字
	consumer1 := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "queue1")
	// 隊(duì)列綁定到exchange
	consumer1.Bind("exchange2", "*.test.*")
	//接收消息
	msgs := consumer1.Consume()
	go func() {
		for d := range msgs {
			log.Printf("Consumer1 received a message: %s", d.Body)
			d.Ack(false)
		}
	}()
	select {}
}

func main() {
	//第一個(gè)參數(shù)指定rabbitmq服務(wù)器的鏈接，第二個(gè)參數(shù)指定創(chuàng)建隊(duì)列的名字
	consumer2 := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "queue2")
	// 隊(duì)列綁定到exchange
	consumer2.Bind("exchange2", "#.test.#")
	//接收消息
	msgs := consumer2.Consume()
	go func() {
		for d := range msgs {
			log.Printf("Consumer2 received a message: %s", d.Body)
			d.Ack(false)
		}
	}()
	select {}
}