fox/rmq/rmq_test.go

package rmq

import (
	"context"
	"fmt"
	"github.com/fox/fox/log"
	"os"
	"os/signal"
	"syscall"
	"testing"
	"time"
)

//const url = "amqp://admin:password@114.132.124.145:5672/vh_game_dev"
//const exchangeName = "test_e"
//const queueName = "test_q"

func initLog() {
	log.Open("test.log", log.DebugL)
}

func TestRabbitmq(t *testing.T) {
	initLog()
	// 加载配置
	rabbitConfig := LoadRabbitMQConfig()

	// 创建连接
	conn, err := NewConnection(rabbitConfig)
	if err != nil {
		log.ErrorF("Failed to create RabbitMQ connection: %v", err)
		return
	}
	defer func() { _ = conn.Close() }()

	// 启动重连监控
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	go conn.Reconnect(ctx)

	// 创建生产者
	producer := NewProducer(conn, rabbitConfig)
	defer func() { _ = producer.Close() }()

	// 创建消费者处理器
	handler := func(ctx context.Context, msg string) error {
		log.InfoF("Processing message: %s", msg)
		// 这里实现具体的业务逻辑
		time.Sleep(1 * time.Second) // 模拟处理时间
		return nil
	}

	// 创建消费者
	consumer := NewConsumer(conn, rabbitConfig, handler)
	defer func() { _ = consumer.Close() }()

	// 启动消费者
	if err = consumer.StartConsuming(ctx); err != nil {
		log.ErrorF("Failed to start consumer: %v", err)
		return
	}

	// 示例：发送测试消息
	go func() {
		time.Sleep(2 * time.Second) // 等待消费者启动
		for i := 0; i < 10; i++ {
			testMessage := fmt.Sprintf("hello world %d", i)
			if err = producer.Publish(ctx, testMessage); err != nil {
				log.ErrorF("Failed to publish test message: %v", err)
			} else {
				log.Info("Test message published successfully")
			}
		}
	}()

	// 等待中断信号
	sigChan := make(chan os.Signal, 1)
	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
	<-sigChan

	log.Info("Shutting down...")
	cancel()
}