turkmentv_sms_transmitter/cmd/consumer/cs.go

package main

import (
	"context"
	"fmt"
	"log"
	"os"
	"os/signal"
	"syscall"

	"smpp-transmitter/pkg/data"
	fl "smpp-transmitter/pkg/logger"
	"smpp-transmitter/pkg/mq"

	"github.com/joho/godotenv"
	"gorm.io/driver/mysql"
	"gorm.io/gorm"
	"gorm.io/gorm/logger"
)

var (
	rabbitMQURL  string
	mysqlDSN     string
	queueName    string
	exchangeName string
	appEnv       string
	routeKey     string
	Log          fl.LogService
)

func init() {
	err := godotenv.Load()
	if err != nil {
		log.Fatal("Error loading .env file")
	}
	// Load environment variables
	rabbitMQURL := os.Getenv("RABBITMQ_URL")
	mysqlDSN := os.Getenv("MYSQL_DSN")
	queueName := os.Getenv("QUEUE_NAME")
	exchangeName := os.Getenv("EXCHABGE_NAME")
	appEnv := os.Getenv("APP_ENV")
	routeKey := os.Getenv("ROUTE_KEY")

	if rabbitMQURL == "" || mysqlDSN == "" || queueName == "" || exchangeName == "" || routeKey == "" {
		log.Fatal("RABBITMQ_URL, MYSQL_DSN, and QUEUE_NAME environment variables must be set")
	}

	Log, err := fl.SetupLogger(appEnv)
	if err != nil {
		log.Fatalf("Failed to set up logger: %v", err)
	}

	Log.Info("SMS Reply consumer service initialized:")
}

func main() {

	// Initialize MySQL database connection
	db, err := gorm.Open(mysql.Open(mysqlDSN), &gorm.Config{
		Logger: logger.Default.LogMode(logger.Info),
	})

	if err != nil {
		Log.Error("Failed to connect to MySQL:", err)
		log.Fatalf("Failed to connect to MySQL: %v", err)
	}

	//initilize rabbitmq
	messageQueue := mq.NewRabbitMQ(rabbitMQURL, exchangeName, queueName, routeKey)

	// Start message processing
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// Run RabbitMQ consumer
	go messageQueue.Consume(ctx, func(ctx context.Context, body []byte) {
		message := &data.Message{}

		Log.Info("Received a raw message", "body", string(body))

		if err := message.Convert(body); err != nil {
			Log.Error("Message cannot be converted", err)
			return
		}

		if err := message.Insert(ctx, db); err != nil {
			Log.Error("Message cannot be Inserted", err)
			return
		}

		fmt.Printf("Message handled %v", string(body))

	})

	// Handle signals for graceful shutdown
	sigs := make(chan os.Signal, 1)
	signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM)
	<-sigs
	log.Println("Shutting down...")
	Log.Info("Shutting down...")
}