go-svc-mqconsumer/functions/utility_functions.go

package functions

import (
	"encoding/json"
	"net/http"
	"time"

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

// HealthCheck RabbitMQ健康检查
func HealthCheck(w http.ResponseWriter, r *http.Request, rabbitFactory *rabbitmq.RabbitMQFactory) {
	w.Header().Set("Content-Type", "application/json")

	// 尝试创建测试通道
	_, err := rabbitFactory.CreateChannel("health_check")
	if err != nil {
		w.WriteHeader(http.StatusServiceUnavailable)
		json.NewEncoder(w).Encode(map[string]interface{}{
			"status": "down",
			"error":  err.Error(),
			"time":   time.Now().Format(time.RFC3339),
		})
		return
	}

	// 清理测试通道
	rabbitFactory.CloseChannel("health_check")

	w.WriteHeader(http.StatusOK)
	json.NewEncoder(w).Encode(map[string]interface{}{
		"status": "up",
		"time":   time.Now().Format(time.RFC3339),
	})
}

// GetQueueInfo 获取队列信息
func GetQueueInfo(rabbitFactory *rabbitmq.RabbitMQFactory, req types.QueueInfoRequest, reqCtx *ctx.RequestContext) *types.QueryResult {
	// 设置默认值
	if req.ChannelName == "" {
		req.ChannelName = "default"
	}

	// 获取通道
	channel, err := rabbitFactory.GetChannel(req.ChannelName)
	if err != nil {
		return &types.QueryResult{
			Success: false,
			Error:   "Failed to get channel: " + err.Error(),
			Time:    time.Now().Format(time.RFC3339),
		}
	}

	// 获取队列信息
	myQueue, err := channel.QueueInspect(req.QueueName)
	if err != nil {
		return &types.QueryResult{
			Success: false,
			Error:   "Failed to inspect queue: " + err.Error(),
			Time:    time.Now().Format(time.RFC3339),
		}
	}

	return &types.QueryResult{
		Success: true,
		Time:    time.Now().Format(time.RFC3339),
		Data: map[string]interface{}{
			"Consumers": myQueue.Consumers,
			"Messages":  myQueue.Messages,
			"Name":      myQueue.Name,
		},
	}
}