// Package mq define message queue operation functions package mq import ( "context" "crypto/tls" "crypto/x509" "encoding/pem" "fmt" "net/url" "os" "sync" "time" "modelRT/config" "modelRT/logger" amqp "github.com/rabbitmq/amqp091-go" "github.com/youmark/pkcs8" ) var ( _globalRabbitMQProxy *RabbitMQProxy rabbitMQOnce sync.Once ) // RabbitMQProxy define stuct of rabbitMQ connection proxy type RabbitMQProxy struct { Conn *amqp.Connection mu sync.Mutex } // RabbitMQCertConf define stuct of rabbitMQ connection certificates config type RabbitMQCertConf struct { serverName string insecureSkipVerify bool clientCert tls.Certificate caCertPool *x509.CertPool } // GetConn define func to return the rabbitMQ connection func GetConn() *amqp.Connection { _globalRabbitMQProxy.mu.Lock() defer _globalRabbitMQProxy.mu.Unlock() return _globalRabbitMQProxy.Conn } // InitRabbitProxy return instance of rabbitMQ connection func InitRabbitProxy(ctx context.Context, rCfg config.RabbitMQConfig) *RabbitMQProxy { amqpURI := generateRabbitMQURI(rCfg) certConf, err := initCertConf(rCfg) if err != nil { logger.Error(ctx, "init rabbitMQ cert config failed", "error", err) panic(err) } rabbitMQOnce.Do(func() { conn := initRabbitMQ(ctx, amqpURI, certConf) _globalRabbitMQProxy = &RabbitMQProxy{Conn: conn} go _globalRabbitMQProxy.handleReconnect(ctx, amqpURI) }) return _globalRabbitMQProxy } // initRabbitMQ return instance of rabbitMQ connection func initRabbitMQ(ctx context.Context, rabbitMQURI string, certConf *RabbitMQCertConf) *amqp.Connection { logger.Info(ctx, "connecting to rabbitMQ server", "rabbit_uri", rabbitMQURI) tlsConfig := &tls.Config{ Certificates: []tls.Certificate{certConf.clientCert}, RootCAs: certConf.caCertPool, InsecureSkipVerify: certConf.insecureSkipVerify, ServerName: certConf.serverName, } conn, err := amqp.DialConfig(rabbitMQURI, amqp.Config{ TLSClientConfig: tlsConfig, SASL: []amqp.Authentication{&amqp.ExternalAuth{}}, Heartbeat: 10 * time.Second, }) if err != nil { logger.Error(ctx, "init rabbitMQ connection failed", "error", err) } defer conn.Close() return conn } func (p *RabbitMQProxy) handleReconnect(ctx context.Context, rabbitMQURI string) { for { closeChan := make(chan *amqp.Error) GetConn().NotifyClose(closeChan) err, ok := <-closeChan if !ok { logger.Info(ctx, "rabbitMQ notify channel closed, stopping solicitor") break } if err != nil { logger.Warn(ctx, "rabbitMQ connection closed by error", "reason", err) } else { logger.Info(ctx, "rabbitMQ connection closed normally (nil err)") } for { time.Sleep(5 * time.Second) newConn, err := amqp.Dial(rabbitMQURI) if err == nil { p.mu.Lock() p.Conn = newConn p.mu.Unlock() logger.Info(ctx, "rabbitMQ reconnected successfully") break } logger.Error(ctx, "rabbitMQ reconnect failed", "err", err) } } } func generateRabbitMQURI(rCfg config.RabbitMQConfig) string { user := url.QueryEscape(rCfg.User) password := url.QueryEscape(rCfg.Password) amqpURI := fmt.Sprintf("amqp://%s:%s@%s:%d/", user, password, rCfg.Host, rCfg.Port, ) return amqpURI } func initCertConf(rCfg config.RabbitMQConfig) (*RabbitMQCertConf, error) { certConf := &RabbitMQCertConf{ insecureSkipVerify: rCfg.InsecureSkipVerify, serverName: rCfg.ServerName, } caCert, err := os.ReadFile(rCfg.CACertPath) if err != nil { return nil, fmt.Errorf("read server ca file failed: %w", err) } caCertPool := x509.NewCertPool() if ok := caCertPool.AppendCertsFromPEM(caCert); !ok { return nil, fmt.Errorf("failed to parse root certificate from %s", rCfg.CACertPath) } certConf.caCertPool = caCertPool certPEM, err := os.ReadFile(rCfg.ClientCertPath) if err != nil { return nil, fmt.Errorf("read client cert file failed: %w", err) } keyData, err := os.ReadFile(rCfg.ClientKeyPath) if err != nil { return nil, fmt.Errorf("read private key file failed: %w", err) } block, _ := pem.Decode(keyData) if block == nil { return nil, fmt.Errorf("failed to decode PEM block from private key") } der, err := pkcs8.ParsePKCS8PrivateKey(block.Bytes, []byte(rCfg.ClientKeyPassword)) if err != nil { return nil, fmt.Errorf("parse password-protected private key failed: %w", err) } privBytes, err := x509.MarshalPKCS8PrivateKey(der) if err != nil { return nil, fmt.Errorf("marshal private key failed: %w", err) } keyPEM := pem.EncodeToMemory(&pem.Block{Type: "PRIVATE KEY", Bytes: privBytes}) clientCert, err := tls.X509KeyPair(certPEM, keyPEM) if err != nil { return nil, fmt.Errorf("create x509 key pair failed: %w", err) } certConf.clientCert = clientCert return certConf, nil }