Auto-gen queue spec should return consistent name

Because a concrete queue specification should always the same name,
otherwise the queue spec is not really a spec, but a random generator.

Author: Zerpet

pkg/rabbitmqamqp/amqp_connection_recovery_test.go

@@ -1038,32 +1038,7 @@ var _ = Describe("Recovery connection test", func() {
 				Expect(msgReceived.Message().GetData()).To(Equal([]byte("hello")))
 				Expect(msgReceived.Accept(ctx)).To(Succeed())
 
-				// Drop connection
-				Eventually(func() error {
-					return testhelper.DropConnectionContainerID(containerId)
-				}).WithTimeout(5*time.Second).WithPolling(400*time.Millisecond).
-					Should(Succeed(), "expected connection to be closed")
-				stateChange := new(StateChanged)
-				Eventually(ch).Within(5 * time.Second).WithPolling(400 * time.Millisecond).
-					Should(Receive(&stateChange))
-				Expect(stateChange.From).To(Equal(&StateOpen{}))
-				Expect(stateChange.To).To(BeAssignableToTypeOf(&StateClosed{}))
-
-				// Receive reconnecting state
-				Eventually(ch).Within(5 * time.Second).WithPolling(400 * time.Millisecond).
-					Should(Receive())
-
-				By("recovering the connection")
-				// Await reconnection
-				Eventually(func() (bool, error) {
-					conn, err := testhelper.GetConnectionByContainerID(containerId)
-					return conn != nil, err
-				}).WithTimeout(6 * time.Second).WithPolling(400 * time.Millisecond).
-					Should(BeTrueBecause("expected connection to be reconnected"))
-				stateChange = new(StateChanged)
-				Eventually(ch).Within(5 * time.Second).WithPolling(400 * time.Millisecond).
-					Should(Receive(&stateChange))
-				Expect(stateChange.To).To(Equal(&StateOpen{}))
+				dropConnectionAndAwaitReconnectionByContainerID(containerId, ch)
 
 				By("publishing and consuming again")
 				// Publish again
@@ -1112,39 +1087,93 @@ var _ = Describe("Recovery connection test", func() {
 				ch := make(chan *StateChanged, 1)
 				conn.NotifyStatusChange(ch)
 
-				// Drop connection
-				Eventually(func() error {
-					return testhelper.DropConnectionContainerID(containerId)
-				}).WithTimeout(5*time.Second).WithPolling(400*time.Millisecond).
-					Should(Succeed(), "expected connection to be closed")
-				stateChange := new(StateChanged)
-				Eventually(ch).Within(5 * time.Second).WithPolling(400 * time.Millisecond).
-					Should(Receive(&stateChange))
-				Expect(stateChange.From).To(Equal(&StateOpen{}))
-				Expect(stateChange.To).To(BeAssignableToTypeOf(&StateClosed{}))
-
-				// Receive reconnecting state
-				Eventually(ch).Within(5 * time.Second).WithPolling(400 * time.Millisecond).
-					Should(Receive())
-
-				By("recovering the connection")
-				// Await reconnection
-				Eventually(func() (bool, error) {
-					conn, err := testhelper.GetConnectionByContainerID(containerId)
-					return conn != nil, err
-				}).WithTimeout(6 * time.Second).WithPolling(400 * time.Millisecond).
-					Should(BeTrueBecause("expected connection to be reconnected"))
-				stateChange = new(StateChanged)
-				Eventually(ch).Within(5 * time.Second).WithPolling(400 * time.Millisecond).
-					Should(Receive(&stateChange))
-				Expect(stateChange.To).To(Equal(&StateOpen{}))
+				dropConnectionAndAwaitReconnectionByContainerID(containerId, ch)
 
 				By("verifying that the recovery records are not duplicated")
 				Expect(conn.topologyRecoveryRecords.queues).To(HaveLen(1))
 				Expect(conn.topologyRecoveryRecords.bindings).To(HaveLen(1))
 
 				_ = conn.Close(ctx)
 			})
+
+			It("should recover auto-generated queues bound to persistent exchanges", func(ctx context.Context) {
+				conn, err := env.NewConnection(ctx)
+				Expect(err).ToNot(HaveOccurred())
+				ch := make(chan *StateChanged, 1)
+				conn.NotifyStatusChange(ch)
+
+				const exchange = "amq.fanout"
+
+				queue, err := conn.Management().DeclareQueue(ctx, &AutoGeneratedQueueSpecification{
+					IsAutoDelete: true,
+					IsExclusive:  true,
+				})
+				Expect(err).ToNot(HaveOccurred())
+
+				_, err = conn.Management().Bind(ctx, &ExchangeToQueueBindingSpecification{
+					SourceExchange:   exchange,
+					DestinationQueue: queue.Name(),
+					BindingKey:       "auto-generated-queue-bound-to-persistent-exchange",
+				})
+				Expect(err).ToNot(HaveOccurred())
+
+				producer, err := conn.NewPublisher(ctx, &ExchangeAddress{Exchange: exchange, Key: "auto-generated-queue-bound-to-persistent-exchange"}, nil)
+				Expect(err).ToNot(HaveOccurred())
+
+				msg := NewMessage([]byte("hello"))
+				result, err := producer.Publish(ctx, msg)
+				Expect(err).ToNot(HaveOccurred())
+				Expect(result.Outcome).To(Equal(&StateAccepted{}))
+
+				consumer, err := conn.NewConsumer(ctx, queue.Name(), nil)
+				Expect(err).ToNot(HaveOccurred())
+
+				msgReceived, err := consumer.Receive(ctx)
+				Expect(err).ToNot(HaveOccurred())
+				Expect(msgReceived.Message().GetData()).To(Equal([]byte("hello")))
+				Expect(msgReceived.Accept(ctx)).To(Succeed())
+
+				dropConnectionAndAwaitReconnectionByContainerID(containerId, ch)
+
+				msg = NewMessage([]byte("hello again"))
+				result, err = producer.Publish(ctx, msg)
+				Expect(err).ToNot(HaveOccurred())
+				Expect(result.Outcome).To(Equal(&StateAccepted{}))
+
+				msgReceived, err = consumer.Receive(ctx)
+				Expect(err).ToNot(HaveOccurred())
+				Expect(msgReceived.Message().GetData()).To(Equal([]byte("hello again")))
+				Expect(msgReceived.Accept(ctx)).To(Succeed())
+			})
 		})
 	})
 })
+
+func dropConnectionAndAwaitReconnectionByContainerID(containerID string, ch <-chan *StateChanged) {
+	// Drop connection
+	Eventually(func() error {
+		return testhelper.DropConnectionContainerID(containerID)
+	}).WithTimeout(5*time.Second).WithPolling(400*time.Millisecond).WithOffset(1).
+		Should(Succeed(), "expected connection to be closed")
+	stateChange := new(StateChanged)
+	Eventually(ch).Within(5 * time.Second).WithPolling(400 * time.Millisecond).WithOffset(1).
+		Should(Receive(&stateChange))
+	Expect(stateChange.From).To(Equal(&StateOpen{}))
+	Expect(stateChange.To).To(BeAssignableToTypeOf(&StateClosed{}))
+
+	// Receive reconnecting state
+	Eventually(ch).Within(5 * time.Second).WithPolling(400 * time.Millisecond).WithOffset(1).
+		Should(Receive())
+
+	By("recovering the connection")
+	// Await reconnection
+	Eventually(func() (bool, error) {
+		conn, err := testhelper.GetConnectionByContainerID(containerID)
+		return conn != nil, err
+	}).WithTimeout(6 * time.Second).WithPolling(400 * time.Millisecond).WithOffset(1).
+		Should(BeTrueBecause("expected connection to be reconnected"))
+	stateChange = new(StateChanged)
+	Eventually(ch).Within(5 * time.Second).WithPolling(400 * time.Millisecond).WithOffset(1).
+		Should(Receive(&stateChange))
+	Expect(stateChange.To).To(Equal(&StateOpen{}))
+}

pkg/rabbitmqamqp/entities.go

@@ -254,6 +254,7 @@ It is a classic queue with auto-generated name.
 It is useful in context like RPC or when you need a temporary queue.
 */
 type AutoGeneratedQueueSpecification struct {
+	genName        string
 	IsAutoDelete   bool
 	IsExclusive    bool
 	MaxLength      int64
@@ -262,7 +263,10 @@ type AutoGeneratedQueueSpecification struct {
 }
 
 func (a *AutoGeneratedQueueSpecification) name() string {
-	return generateNameWithDefaultPrefix()
+	if a.genName == "" {
+		a.genName = generateNameWithDefaultPrefix()
+	}
+	return a.genName
 }
 
 func (a *AutoGeneratedQueueSpecification) isAutoDelete() bool {

pkg/rabbitmqamqp/entities_test.go

@@ -0,0 +1,22 @@
+package rabbitmqamqp
+
+import (
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+var _ = Describe("Entities", func() {
+	Describe("AutoGeneratedQueueSpecification", func() {
+		It("should generate a unique name consistently", func() {
+			queue := &AutoGeneratedQueueSpecification{}
+			generatedName := queue.name()
+			Expect(queue.name()).To(Equal(generatedName), "same instance should generate the same name")
+
+			anotherQueue := &AutoGeneratedQueueSpecification{}
+			anotherGeneratedName := anotherQueue.name()
+			Expect(anotherQueue.name()).To(Equal(anotherGeneratedName), "same instance should generate the same name")
+
+			Expect(generatedName).ToNot(Equal(anotherGeneratedName), "different instances should generate different names")
+		})
+	})
+})