zyos_go_mq/test/single_queue_test.go

package test

import (
	"code.fnuoos.com/go_rely_warehouse/zyos_go_mq.git/rabbit"
	"code.fnuoos.com/go_rely_warehouse/zyos_go_mq.git/rabbitmq"
	"fmt"
	"github.com/streadway/amqp"
	"log"
	"strconv"
	"strings"
	"testing"
)

const SingleQueueName = "test_single_queue"

func TestSend(t *testing.T) {
	var host, port, user, pwd string
	log.SetFlags(log.Lshortfile | log.LstdFlags)
	log.Println("input host,port,user,pwd")
	fmt.Scanf("%s %s %s %s", &host, &port, &user, &pwd)
	if host == "" && port == "" && user == "" && pwd == "" {
		host = "119.23.182.117"
		port = "5672"
		user = "admin"
		pwd = "123456"
	}
	rabbitmq.Init(host, port, user, pwd)
	//第一个参数指定rabbitmq服务器的链接，第二个参数指定创建队列的名字
	sendMq := rabbitmq.New(rabbitmq.Cfg.Uri, SingleQueueName)
	i := 0
	for {
		if i > 99 {
			break
		}
		//time.Sleep(time.Second * 1)
		greetings := []string{"HelloWorld!", strconv.Itoa(i)}
		sendMq.Send(SingleQueueName, strings.Join(greetings, " "))
		i = i + 1
	}
}

func TestReceive(t *testing.T) {
	var host, port, user, pwd string
	log.SetFlags(log.Lshortfile | log.LstdFlags)
	log.Println("input host,port,user,pwd")
	fmt.Scanf("%s %s %s %s", &host, &port, &user, &pwd)
	if host == "" && port == "" && user == "" && pwd == "" {
		host = "119.23.182.117"
		port = "5672"
		user = "admin"
		pwd = "123456"
	}
	//ch, err := GetChannel
	rabbitmq.Init(host, port, user, pwd)
	//第一个参数指定rabbitmq服务器的链接，第二个参数指定创建队列的名字
	receiveMq := rabbitmq.New(rabbitmq.Cfg.Uri, SingleQueueName)
	//for {
	receiveMq.Qos(1)
	//接收消息时，指定
	messages := receiveMq.Consume()

	forever := make(chan bool) //开辟管道
	var i = 0
	//启用协程处理消息
	go func() {
		for msg := range messages {
			log.Printf("Received a message: %s", msg.Body)
			i++
			if i >= 5 {
				msg.Reject(true)
				receiveMq.CloseChannel()
				return
			} else {
				msg.Ack(false)
			}
		}
	}()
	log.Printf("[*] Waiting for message,To exit press CTRL+C")
	<-forever
}

func TestReceive1(t *testing.T) {
	var host, port, user, pwd string
	log.SetFlags(log.Lshortfile | log.LstdFlags)
	log.Println("input host,port,user,pwd")
	fmt.Scanf("%s %s %s %s", &host, &port, &user, &pwd)
	if host == "" && port == "" && user == "" && pwd == "" {
		host = "119.23.182.117"
		port = "5672"
		user = "admin"
		pwd = "123456"
	}
	rabbit.Init(host, port, user, pwd)
	ch, err := rabbit.Cfg.Pool.GetChannel()
	if err != nil {
		fmt.Println(err.Error())
	}

	defer ch.Release()
	ch.Qos(5)
	delivery := ch.Consume(SingleQueueName)

	var res amqp.Delivery
	var ok bool
	//for {
	res, ok = <-delivery
	if ok == true {
		fmt.Println(string(res.Body))
		_ = res.Ack(false)
	} else {
		fmt.Println("error getting message")
		//break
	}
	//}
	fmt.Println("get msg done")
}

func TestReceive2(t *testing.T) {
	var host, port, user, pwd string
	log.SetFlags(log.Lshortfile | log.LstdFlags)
	log.Println("input host,port,user,pwd")
	fmt.Scanf("%s %s %s %s", &host, &port, &user, &pwd)
	if host == "" && port == "" && user == "" && pwd == "" {
		host = "119.23.182.117"
		port = "5672"
		user = "admin"
		pwd = "123456"
	}
	//ch, err := GetChannel
	rabbitmq.Init(host, port, user, pwd)
	//第一个参数指定rabbitmq服务器的链接，第二个参数指定创建队列的名字
	receiveMq := rabbitmq.New(rabbitmq.Cfg.Uri, SingleQueueName)
	//for {
	receiveMq.Qos(1)
	//接收消息时，指定
	messages, ok, err := receiveMq.Get()
	if err != nil {
		t.Fatalf("Failed get: %v", err)
	}

	if !ok {
		t.Fatalf("Get on a queued message did not find the message")
	}
	fmt.Println(">>>>>>ok>>>>>>", ok)
	fmt.Println(">>>>>>messages>>>>>>", string(messages.Body))
}