go-svc-mqconsumer/main.go

package main

import (
	"fmt"
	"log"
	"net/http"
	"os"
	"os/signal"
	"syscall"

	"git.x2erp.com/qdy/go-base/config"
	"git.x2erp.com/qdy/go-base/middleware"
	"git.x2erp.com/qdy/go-base/myservice"
	"git.x2erp.com/qdy/go-db/factory/rabbitmq"
	"git.x2erp.com/qdy/go-db/myhandle"
	"git.x2erp.com/qdy/go-svc-mqconsumer/functions"
	"go-micro.dev/v4/web"
)

func main() {
	cfg, err := config.GetConfig()
	if err != nil {
		log.Fatalf("Failed to create RabbitMQ factory: %v", err)
		return
	}

	serviceConfig := cfg.GetService()

	log.Printf("RabbitMQ Service Starting...")
	log.Printf("Service Port: %d", serviceConfig.Port)
	log.Printf("Service Name: %s", serviceConfig.ServiceName)

	// 启动微服务
	startRabbitMQService(cfg)
}

// 启动RabbitMQ微服务
func startRabbitMQService(cfg config.IConfig) {

	// 初始化RabbitMQ工厂
	rabbitFactory, err := rabbitmq.NewRabbitMQFactory()
	if err != nil {
		log.Fatalf("Failed to create RabbitMQ factory: %v", err)
	}
	defer func() {
		if err := rabbitFactory.Close(); err != nil {
			log.Printf("RabbitMQ close error: %v", err)
		}
	}()

	// 设置优雅关闭
	setupGracefulShutdown(rabbitFactory)

	// 创建默认通道
	if err := createDefaultChannel(rabbitFactory); err != nil {
		log.Fatalf("Failed to create default channel: %v", err)
	}

	webService := myservice.StartStandalone(cfg)

	// 注册HTTP路由到webService
	registerRabbitMQRoutes(webService, rabbitFactory)

	// 等待服务运行
	log.Printf("RabbitMQ Service started successfully")
	log.Printf(" 	• Host: %s", cfg.GetRabbitMQ().Host)
	log.Printf(" 	• Port: %d", cfg.GetRabbitMQ().Port)
	log.Printf(" 	• Vhost: %s", cfg.GetRabbitMQ().Vhost)

	// 保持主程序运行
	select {}
}

// 创建默认通道
func createDefaultChannel(rabbitFactory *rabbitmq.RabbitMQFactory) error {
	_, err := rabbitFactory.CreateChannel("default")
	if err != nil {
		return fmt.Errorf("failed to create default channel: %v", err)
	}
	log.Println("Default RabbitMQ channel created successfully")
	return nil
}

// 注册RabbitMQ相关路由
func registerRabbitMQRoutes(webService web.Service, rabbitFactory *rabbitmq.RabbitMQFactory) {
	// 创建交换机
	webService.Handle("/api/rabbitmq/consumers/stop", middleware.JWTAuthMiddleware(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		myhandle.QueryHandlerJson(w, r, rabbitFactory, functions.CreateExchange)
	})))

	// 创建队列
	webService.Handle("/api/rabbitmq/consumers/start", middleware.JWTAuthMiddleware(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		myhandle.QueryHandlerJson(w, r, rabbitFactory, functions.CreateQueue)
	})))

	// 绑定队列到交换机
	webService.Handle("/api/rabbitmq/queue/bind", middleware.JWTAuthMiddleware(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		myhandle.QueryHandlerJson(w, r, rabbitFactory, functions.BindQueue)
	})))

	// 发送JSON消息
	webService.Handle("/api/rabbitmq/message/send", middleware.JWTAuthMiddleware(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		myhandle.QueryHandlerJson(w, r, rabbitFactory, functions.SendMessage)
	})))

	// 发送原始消息（字节流）
	webService.Handle("/api/rabbitmq/message/send/bytes", middleware.JWTAuthMiddleware(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		myhandle.QueryHandlerJson(w, r, rabbitFactory, functions.SendBytesMessage)
	})))

	// 健康检查
	webService.Handle("/api/rabbitmq/health", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		functions.HealthCheck(w, r, rabbitFactory)
	}))

	// 获取队列信息
	webService.Handle("/api/rabbitmq/queue/info", middleware.JWTAuthMiddleware(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		myhandle.QueryHandlerJson(w, r, rabbitFactory, functions.GetQueueInfo)
	})))
}

// 设置优雅关闭
func setupGracefulShutdown(rabbitFactory *rabbitmq.RabbitMQFactory) {
	signalCh := make(chan os.Signal, 1)
	signal.Notify(signalCh, os.Interrupt, syscall.SIGTERM)

	go func() {
		<-signalCh
		log.Println("\nReceived shutdown signal, closing RabbitMQ connections...")
		if err := rabbitFactory.Close(); err != nil {
			log.Printf("Error closing RabbitMQ: %v", err)
		}
		log.Println("RabbitMQ connections closed gracefully")
		os.Exit(0)
	}()
}