package eventbus import ( "context" "fmt" "time" amqp "github.com/rabbitmq/amqp091-go" "go.uber.org/zap" ) // RabbitMQConfig contient la configuration pour RabbitMQ type RabbitMQConfig struct { URL string MaxRetries int RetryInterval time.Duration Enable bool // Si false, l'EventBus sera désactivé } // EventBusUnavailableError est retourné si l'EventBus est désactivé ou non disponible type EventBusUnavailableError struct { Msg string } func (e *EventBusUnavailableError) Error() string { return e.Msg } // RabbitMQEventBus gère la connexion et les opérations RabbitMQ type RabbitMQEventBus struct { conn *amqp.Connection channel *amqp.Channel config *RabbitMQConfig logger *zap.Logger IsEnabled bool // Indique si l'EventBus est actif } // NewRabbitMQEventBusWithRetry initialise une connexion RabbitMQ avec retry func NewRabbitMQEventBusWithRetry(cfg *RabbitMQConfig, logger *zap.Logger) (*RabbitMQEventBus, error) { if !cfg.Enable { logger.Info("📴 EventBus RabbitMQ désactivé par configuration.") return &RabbitMQEventBus{config: cfg, logger: logger, IsEnabled: false}, nil } if cfg.MaxRetries == 0 { cfg.MaxRetries = 1 } if cfg.RetryInterval == 0 { cfg.RetryInterval = 5 * time.Second } var conn *amqp.Connection var err error for i := 0; i < cfg.MaxRetries; i++ { logger.Info("🔄 Tentative de connexion à RabbitMQ", zap.Int("attempt", i+1), zap.Int("max_attempts", cfg.MaxRetries), zap.String("url", cfg.URL)) conn, err = amqp.Dial(cfg.URL) if err == nil { logger.Info("✅ Connexion à RabbitMQ établie avec succès.") channel, err := conn.Channel() if err != nil { conn.Close() return nil, fmt.Errorf("failed to open RabbitMQ channel: %w", err) } return &RabbitMQEventBus{conn: conn, channel: channel, config: cfg, logger: logger, IsEnabled: true}, nil } logger.Warn("❌ Échec de connexion à RabbitMQ", zap.Error(err), zap.Int("attempt", i+1), zap.Int("max_attempts", cfg.MaxRetries)) if i < cfg.MaxRetries-1 { logger.Info("🔄 Nouvelle tentative de connexion RabbitMQ dans quelques secondes...", zap.Duration("interval", cfg.RetryInterval)) time.Sleep(cfg.RetryInterval) } } // Si toutes les tentatives échouent, décider du mode dégradé ou fatal logger.Error("❌ Échec de connexion à RabbitMQ après toutes les tentatives.", zap.Int("max_attempts", cfg.MaxRetries), zap.Error(err)) return nil, &EventBusUnavailableError{Msg: fmt.Sprintf("failed to connect to RabbitMQ after %d attempts: %v", cfg.MaxRetries, err)} } // Publish envoie un message à un exchange RabbitMQ func (eb *RabbitMQEventBus) Publish(ctx context.Context, exchange, routingKey string, mandatory, immediate bool, msg amqp.Publishing) error { if !eb.IsEnabled { eb.logger.Warn("⚠️ Tentative de publication sur EventBus désactivé", zap.String("exchange", exchange), zap.String("routing_key", routingKey)) return &EventBusUnavailableError{Msg: "EventBus is disabled"} } return eb.channel.PublishWithContext(ctx, exchange, routingKey, mandatory, immediate, msg) } // Consume démarre un consommateur RabbitMQ func (eb *RabbitMQEventBus) Consume(queue, consumer string, autoAck, exclusive, noLocal, noWait bool, args amqp.Table) (<-chan amqp.Delivery, error) { if !eb.IsEnabled { eb.logger.Warn("⚠️ Tentative de consommation sur EventBus désactivé", zap.String("queue", queue), zap.String("consumer", consumer)) return nil, &EventBusUnavailableError{Msg: "EventBus is disabled"} } return eb.channel.Consume(queue, consumer, autoAck, exclusive, noLocal, noWait, args) } // Close ferme la connexion et le canal RabbitMQ func (eb *RabbitMQEventBus) Close() error { if !eb.IsEnabled { return nil } var errs []error if eb.channel != nil { if err := eb.channel.Close(); err != nil { errs = append(errs, fmt.Errorf("failed to close RabbitMQ channel: %w", err)) } } if eb.conn != nil { if err := eb.conn.Close(); err != nil { errs = append(errs, fmt.Errorf("failed to close RabbitMQ connection: %w", err)) } } if len(errs) > 0 { return fmt.Errorf("errors closing RabbitMQ: %v", errs) } eb.logger.Info("🔌 Connexion RabbitMQ fermée.") return nil } // Health vérifie la santé de la connexion RabbitMQ func (eb *RabbitMQEventBus) Health() error { if !eb.IsEnabled { return fmt.Errorf("RabbitMQ EventBus est désactivé") } if eb.conn == nil || eb.conn.IsClosed() { return fmt.Errorf("connexion RabbitMQ non établie ou fermée") } // Tenter d'ouvrir un canal temporaire pour vérifier l'état de la connexion tmpChannel, err := eb.conn.Channel() if err != nil { return fmt.Errorf("impossible d'ouvrir un canal RabbitMQ: %w", err) } tmpChannel.Close() // Fermer le canal temporaire return nil }