Add topology recovery infrastructure and configuration options

This commit introduces the foundational infrastructure for topology recovery:
- Add TopologyRecoveryOptions configuration with three modes: disabled, transient queues only, and full recovery
- Implement topologyRecoveryRecords to track queues, exchanges, and bindings
- Add recovery record types (queueRecoveryRecord, exchangeRecoveryRecord, bindingRecoveryRecord) with conversion methods to specifications
- Integrate topology recovery tracking into AmqpConnection and AmqpManagement
- Add helper method isQueueDestinationForBindingTransient to determine if binding destinations are transient
- Include comprehensive unit tests for recovery record conversions and transient queue detection

This is a first step to support topology recovery. The actual recovery logic will be implemented in subsequent commits.

Author: Zerpet

pkg/rabbitmqamqp/amqp_connection.go

@@ -45,6 +45,20 @@ func (o OAuth2Options) Clone() *OAuth2Options {
 
 }
 
+// TopologyRecoveryOptions is used to configure the topology recovery behavior of the connection.
+// See [TopologyRecoveryDisabled], [TopologyRecoveryOnlyTransientQueues], and [TopologyRecoveryAllEnabled] for more information.
+type TopologyRecoveryOptions byte
+
+const (
+	// TopologyRecoveryDisabled disables the topology recovery.
+	TopologyRecoveryDisabled TopologyRecoveryOptions = iota
+	// TopologyRecoveryOnlyTransientQueues recovers only queues declared as exclusive and/or auto delete, and
+	// related bindings. Exchanges are not recovered.
+	TopologyRecoveryOnlyTransientQueues
+	// TopologyRecoveryAllEnabled recovers all the topology. All exchanges, queues, and bindings are recovered.
+	TopologyRecoveryAllEnabled
+)
+
 type AmqpConnOptions struct {
 	// wrapper for amqp.ConnOptions
 	ContainerID string
@@ -76,6 +90,9 @@ type AmqpConnOptions struct {
 	// when the connection is closed unexpectedly.
 	RecoveryConfiguration *RecoveryConfiguration
 
+	// TopologyRecoveryOptions is used to configure the topology recovery behavior of the connection.
+	TopologyRecoveryOptions TopologyRecoveryOptions
+
 	// The OAuth2Options is used to configure the connection with OAuth2 token.
 	OAuth2Options *OAuth2Options
 
@@ -116,16 +133,17 @@ type AmqpConnection struct {
 	properties        map[string]any
 	featuresAvailable *featuresAvailable
 
-	azureConnection *amqp.Conn
-	management      *AmqpManagement
-	lifeCycle       *LifeCycle
-	amqpConnOptions *AmqpConnOptions
-	address         string
-	session         *amqp.Session
-	refMap          *sync.Map
-	entitiesTracker *entitiesTracker
-	mutex           sync.RWMutex
-	closed          bool
+	azureConnection         *amqp.Conn
+	management              *AmqpManagement
+	lifeCycle               *LifeCycle
+	amqpConnOptions         *AmqpConnOptions
+	address                 string
+	session                 *amqp.Session
+	refMap                  *sync.Map
+	entitiesTracker         *entitiesTracker
+	topologyRecoveryRecords *topologyRecoveryRecords
+	mutex                   sync.RWMutex
+	closed                  bool
 }
 
 func (a *AmqpConnection) Properties() map[string]any {
@@ -321,16 +339,19 @@ func Dial(ctx context.Context, address string, connOptions *AmqpConnOptions) (*A
 	if err != nil {
 		return nil, err
 	}
-
 	// create the connection
 	conn := &AmqpConnection{
-		management:        newAmqpManagement(),
-		lifeCycle:         NewLifeCycle(),
-		amqpConnOptions:   connOptions,
-		entitiesTracker:   newEntitiesTracker(),
-		featuresAvailable: newFeaturesAvailable(),
+		management:              newAmqpManagement(connOptions.TopologyRecoveryOptions),
+		lifeCycle:               NewLifeCycle(),
+		amqpConnOptions:         connOptions,
+		entitiesTracker:         newEntitiesTracker(),
+		topologyRecoveryRecords: newTopologyRecoveryRecords(),
+		featuresAvailable:       newFeaturesAvailable(),
 	}
 
+	// management needs to access the connection to manage the recovery records
+	conn.management.topologyRecoveryRecords = conn.topologyRecoveryRecords
+
 	err = conn.open(ctx, address, connOptions)
 	if err != nil {
 		return nil, err
@@ -497,6 +518,7 @@ func (a *AmqpConnection) maybeReconnect() {
 
 		if err == nil {
 			a.restartEntities()
+			// TODO: integration point for topology recovery
 			a.lifeCycle.SetState(&StateOpen{})
 			return
 		}

pkg/rabbitmqamqp/amqp_connection_recovery.go

@@ -2,6 +2,7 @@ package rabbitmqamqp
 
 import (
 	"errors"
+	"slices"
 	"sync"
 	"time"
 )
@@ -90,3 +91,159 @@ func (e *entitiesTracker) CleanUp() {
 		return true
 	})
 }
+
+type queueRecoveryRecord struct {
+	queueName  string
+	queueType  TQueueType
+	autoDelete *bool
+	exclusive  *bool
+	arguments  map[string]any
+}
+
+func (q *queueRecoveryRecord) toIQueueSpecification() IQueueSpecification {
+	switch q.queueType {
+	case Quorum:
+		return &QuorumQueueSpecification{
+			Name:      q.queueName,
+			Arguments: q.arguments,
+		}
+	case Classic:
+		return &ClassicQueueSpecification{
+			Name:         q.queueName,
+			IsAutoDelete: *q.autoDelete,
+			IsExclusive:  *q.exclusive,
+			Arguments:    q.arguments,
+		}
+	case Stream:
+		return &StreamQueueSpecification{
+			Name:      q.queueName,
+			Arguments: q.arguments,
+		}
+	}
+	return nil
+}
+
+type exchangeRecoveryRecord struct {
+	exchangeName string
+	exchangeType TExchangeType
+	autoDelete   bool
+	arguments    map[string]any
+}
+
+func (e *exchangeRecoveryRecord) toIExchangeSpecification() IExchangeSpecification {
+	switch e.exchangeType {
+	case Direct:
+		return &DirectExchangeSpecification{
+			Name:         e.exchangeName,
+			IsAutoDelete: e.autoDelete,
+			Arguments:    e.arguments,
+		}
+	case Topic:
+		return &TopicExchangeSpecification{
+			Name:         e.exchangeName,
+			IsAutoDelete: e.autoDelete,
+			Arguments:    e.arguments,
+		}
+	case FanOut:
+		return &FanOutExchangeSpecification{
+			Name:         e.exchangeName,
+			IsAutoDelete: e.autoDelete,
+			Arguments:    e.arguments,
+		}
+	case Headers:
+		return &HeadersExchangeSpecification{
+			Name:         e.exchangeName,
+			IsAutoDelete: e.autoDelete,
+			Arguments:    e.arguments,
+		}
+	default:
+		return &CustomExchangeSpecification{
+			Name:             e.exchangeName,
+			IsAutoDelete:     e.autoDelete,
+			ExchangeTypeName: string(e.exchangeType),
+			Arguments:        e.arguments,
+		}
+	}
+}
+
+type bindingRecoveryRecord struct {
+	sourceExchange     string
+	destination        string
+	isDestinationQueue bool
+	bindingKey         string
+	arguments          map[string]any
+	path               string
+}
+
+func (b *bindingRecoveryRecord) toIBindingSpecification() IBindingSpecification {
+	if b.isDestinationQueue {
+		return &ExchangeToQueueBindingSpecification{
+			SourceExchange:   b.sourceExchange,
+			DestinationQueue: b.destination,
+			BindingKey:       b.bindingKey,
+			Arguments:        b.arguments,
+		}
+	}
+	return &ExchangeToExchangeBindingSpecification{
+		SourceExchange:      b.sourceExchange,
+		DestinationExchange: b.destination,
+		BindingKey:          b.bindingKey,
+		Arguments:           b.arguments,
+	}
+}
+
+type topologyRecoveryRecords struct {
+	queues    []queueRecoveryRecord
+	exchanges []exchangeRecoveryRecord
+	bindings  []bindingRecoveryRecord
+}
+
+func newTopologyRecoveryRecords() *topologyRecoveryRecords {
+	return &topologyRecoveryRecords{
+		queues:    make([]queueRecoveryRecord, 0),
+		exchanges: make([]exchangeRecoveryRecord, 0),
+		bindings:  make([]bindingRecoveryRecord, 0),
+	}
+}
+
+func (t *topologyRecoveryRecords) addQueueRecord(record queueRecoveryRecord) {
+	t.queues = append(t.queues, record)
+}
+
+func (t *topologyRecoveryRecords) removeQueueRecord(record queueRecoveryRecord) {
+	t.queues = slices.DeleteFunc(t.queues, func(r queueRecoveryRecord) bool {
+		return r.queueName == record.queueName
+	})
+}
+
+func (t *topologyRecoveryRecords) addExchangeRecord(record exchangeRecoveryRecord) {
+	t.exchanges = append(t.exchanges, record)
+}
+
+func (t *topologyRecoveryRecords) removeExchangeRecord(record exchangeRecoveryRecord) {
+	t.exchanges = slices.DeleteFunc(t.exchanges, func(r exchangeRecoveryRecord) bool {
+		return r.exchangeName == record.exchangeName
+	})
+}
+
+func (t *topologyRecoveryRecords) addBindingRecord(record bindingRecoveryRecord) {
+	t.bindings = append(t.bindings, record)
+}
+
+func (t *topologyRecoveryRecords) removeBindingRecord(bindingPath string) {
+	t.bindings = slices.DeleteFunc(t.bindings, func(r bindingRecoveryRecord) bool {
+		return r.path == bindingPath
+	})
+}
+
+func (t *topologyRecoveryRecords) removeBindingRecordBySourceExchange(sourceExchange string) {
+	t.bindings = slices.DeleteFunc(t.bindings, func(r bindingRecoveryRecord) bool {
+		return r.sourceExchange == sourceExchange
+	})
+}
+
+func (t *topologyRecoveryRecords) removeBindingRecordByDestinationQueue(destinationQueue string) {
+	t.bindings = slices.DeleteFunc(t.bindings, func(r bindingRecoveryRecord) bool {
+		return r.destination == destinationQueue
+	})
+}