go-svc-mqconsumer/functions/queue_functions.go

package functions

import (
	"time"

	"git.x2erp.com/qdy/go-base/ctx"
	"git.x2erp.com/qdy/go-base/types"
	"git.x2erp.com/qdy/go-db/factory/rabbitmq"
)

// CreateQueue 创建队列
func CreateQueue(rabbitFactory *rabbitmq.RabbitMQFactory, req types.QueueRequest, reqCtx *ctx.RequestContext) *types.QueryResult {
	// 设置默认值
	if req.ChannelName == "" {
		req.ChannelName = "default"
	}

	// 确保通道存在
	_, err := rabbitFactory.CreateChannel(req.ChannelName)
	if err != nil {
		return &types.QueryResult{
			Success: false,
			Error:   "Failed to create channel: " + err.Error(),
			Time:    time.Now().Format(time.RFC3339),
		}
	}

	// 创建队列
	err = rabbitFactory.AddQueue(
		req.ChannelName,
		req.QueueName,
		req.Durable,
		req.Exclusive,
		req.AutoDelete,
	)
	if err != nil {
		return &types.QueryResult{
			Success: false,
			Error:   "Failed to create queue: " + err.Error(),
			Time:    time.Now().Format(time.RFC3339),
		}
	}

	return &types.QueryResult{
		Success: true,
		Message: "Queue created successfully",
		Time:    time.Now().Format(time.RFC3339),
		Data: map[string]interface{}{
			"queue_name":  req.QueueName,
			"durable":     req.Durable,
			"exclusive":   req.Exclusive,
			"auto_delete": req.AutoDelete,
			"channel":     req.ChannelName,
		},
	}
}

// BindQueue 绑定队列到交换机
func BindQueue(rabbitFactory *rabbitmq.RabbitMQFactory, req types.QueueBindRequest, reqCtx *ctx.RequestContext) *types.QueryResult {
	// 设置默认值
	if req.ChannelName == "" {
		req.ChannelName = "default"
	}
	if req.RoutingKey == "" {
		req.RoutingKey = req.QueueName
	}

	// 绑定队列
	err := rabbitFactory.BindQueue(
		req.ChannelName,
		req.QueueName,
		req.ExchangeName,
		req.RoutingKey,
	)
	if err != nil {
		return &types.QueryResult{
			Success: false,
			Error:   "Failed to bind queue: " + err.Error(),
			Time:    time.Now().Format(time.RFC3339),
		}
	}

	return &types.QueryResult{
		Success: true,
		Message: "Queue bound to exchange successfully",
		Time:    time.Now().Format(time.RFC3339),
		Data: map[string]interface{}{
			"queue":       req.QueueName,
			"exchange":    req.ExchangeName,
			"routing_key": req.RoutingKey,
			"channel":     req.ChannelName,
		},
	}
}