Fix race condition in AmqpConnection

Author: Zerpet

pkg/rabbitmqamqp/amqp_connection.go

@@ -121,6 +121,8 @@ type AmqpConnection struct {
 	session         *amqp.Session
 	refMap          *sync.Map
 	entitiesTracker *entitiesTracker
+	mutex           sync.RWMutex
+	closed          bool
 }
 
 func (a *AmqpConnection) Properties() map[string]any {
@@ -456,6 +458,13 @@ func (a *AmqpConnection) open(ctx context.Context, address string, connOptions *
 	return nil
 }
 func (a *AmqpConnection) maybeReconnect() {
+	a.mutex.RLock()
+	isClosed := a.closed
+	a.mutex.RUnlock()
+	if isClosed {
+		Info("Connection is closed, not reconnecting", "ID", a.Id())
+		return
+	}
 
 	if !a.amqpConnOptions.RecoveryConfiguration.ActiveRecovery {
 		Info("Recovery is disabled, closing connection", "ID", a.Id())
@@ -467,6 +476,13 @@ func (a *AmqpConnection) maybeReconnect() {
 	maxDelay := 1 * time.Minute
 
 	for attempt := 1; attempt <= a.amqpConnOptions.RecoveryConfiguration.MaxReconnectAttempts; attempt++ {
+		a.mutex.RLock()
+		isClosed := a.closed
+		a.mutex.RUnlock()
+		if isClosed {
+			Info("Connection is closed, aborting reconnection attempts", "ID", a.Id())
+			return
+		}
 
 		///wait for before reconnecting
 		// add some random milliseconds to the wait time to avoid thundering herd
@@ -548,6 +564,13 @@ Close closes the connection to the AMQP 1.0 server and the management interface.
 All the publishers and consumers are closed as well.
 */
 func (a *AmqpConnection) Close(ctx context.Context) error {
+	a.mutex.Lock()
+	if a.closed {
+		a.mutex.Unlock()
+		return nil
+	}
+	a.closed = true
+	defer a.mutex.Unlock()
 	// the status closed (lifeCycle.SetState(&StateClosed{error: nil})) is not set here
 	// it is set in the connection.Done() channel
 	// the channel is called anyway