ProjectOctopus/agent-go/rabbitmq/RabbitMQConnector.go

package rabbitmq

import (
	"agent-go/g"
	"fmt"
	"github.com/streadway/amqp"
	"strings"
	"sync"
)

var log = g.G.LOG

// RabbitMQConn is a struct that holds the connection and channel objects
type RabbitMQConn struct {
	Connection *amqp.Connection
	Channel    *amqp.Channel
}

type ConnectProperty struct {
	ExchangeName string
	QueueName    string
	ExchangeType string
	TopicKey     string
}

// 定义全局唯一的 Singleton 实例
var instance *amqp.Connection

// 用 sync.Once 变量确保初始化函数只会被调用一次
var once sync.Once

// 初始化 Singleton 实例的函数
func createInstance() {
	// 在这里进行 Singleton 的初始化操作

	// 获取RabbitMQ的连接地址
	rabbitMQEndpointFromG := parseRabbitMQEndpointFromG()

	// 创建全局唯一连接 RabbitMQ连接
	connection, err := amqp.Dial(rabbitMQEndpointFromG)
	if err != nil {
		log.Error(fmt.Sprintf("failed to connect to RabbitMQ: %v", err))
	}

	instance = connection
}

// GetInstance 获取全局唯一的 Singleton 实例的函数
func GetInstance() *amqp.Connection {
	// 使用 sync.Once 确保 createInstance 只会被调用一次
	once.Do(createInstance)
	return instance
}

// NewRabbitMQConn creates a new RabbitMQ connection object
func NewRabbitMQConn(property *ConnectProperty) (*RabbitMQConn, error) {

	// 获取RabbitMQ的连接
	conn := GetInstance()

	ch, err := conn.Channel()
	if err != nil {
		return nil, fmt.Errorf("failed to create RabbitMQ channel: %w", err)
	}

	if err = ch.ExchangeDeclare(
		property.ExchangeName, // name of the exchange
		property.ExchangeType, // type of the exchange
		true,                  // durable
		false,                 // delete when complete
		false,                 // internal
		false,                 // noWait
		nil,                   // arguments
	); err != nil {
		return nil, fmt.Errorf("failed to declare RabbitMQ exchange: %w", err)
	}

	_, err = ch.QueueDeclare(
		property.QueueName, // name of the queue
		true,               // durable
		false,              // delete when unused
		false,              // exclusive
		false,              // noWait
		nil,                // arguments
	)
	if err != nil {
		return nil, fmt.Errorf("failed to declare RabbitMQ queue: %w", err)
	}

	if err = ch.QueueBind(
		property.QueueName,    // name of the queue
		property.TopicKey,     // routing key - all topics
		property.ExchangeName, // name of the exchange
		false,                 // noWait
		nil,                   // arguments
	); err != nil {
		return nil, fmt.Errorf("failed to bind RabbitMQ queue: %w", err)
	}

	return &RabbitMQConn{Connection: conn, Channel: ch}, nil
}

// parseRabbitMQEndpoint 根据全局变量NacosConfig解析出RabbitMQ的连接地址
func parseRabbitMQEndpointFromG() string {

	nacosConfig := g.G.NacosConfig

	var res strings.Builder

	host := nacosConfig.GetString("spring.rabbitmq.host")
	port := nacosConfig.GetString("spring.rabbitmq.port")
	username := nacosConfig.GetString("spring.rabbitmq.username")
	password := nacosConfig.GetString("spring.rabbitmq.password")
	virtualHost := nacosConfig.GetString("spring.rabbitmq.virtual-host")

	// amqp://{username}:{password}@{hostname}:{port}/{virtual_host}
	res.WriteString("amqp://")
	res.WriteString(username)
	res.WriteString(":")
	res.WriteString(password)
	res.WriteString("@")
	res.WriteString(host)
	res.WriteString(":")
	res.WriteString(port)
	res.WriteString(virtualHost)

	s := res.String()
	log.Debug(fmt.Sprintf("generate RabbitMQ endpoint is %s", s))
	return s
}

func CloseChannel(conn *RabbitMQConn) error {
	var err error

	if conn.Channel != nil {
		if err = conn.Channel.Close(); err != nil {
			log.Error(fmt.Sprintf("Failed to close RabbitMQ channel: %v", err))
		}
	}
	return err
}

// CloseRabbitMQAll closes the RabbitMQ connection and channel
func (r *RabbitMQConn) CloseRabbitMQAll() error {
	var err error

	if r.Channel != nil {
		if err = r.Channel.Close(); err != nil {
			log.Error(fmt.Sprintf("Failed to close RabbitMQ channel: %v", err))
		}
	}

	if r.Connection != nil {
		if err = r.Connection.Close(); err != nil {
			log.Error(fmt.Sprintf("Failed to close RabbitMQ connection: %v", err))
		}
	}

	return err
}