add tests

Signed-off-by: Gabriele Santomaggio <[email protected]>

Author: Gsantomaggio

pkg/rabbitmq_amqp/amqp_connection.go

@@ -136,6 +136,7 @@ func Dial(ctx context.Context, addresses []string, connOptions *AmqpConnOptions,
 	}
 	conn.amqpConnOptions = connOptions
 	conn.amqpConnOptions.addresses = addresses
+	conn.lifeCycle.SetState(&StateOpen{})
 	return conn, nil
 
 }
@@ -204,6 +205,7 @@ func (a *AmqpConnection) open(ctx context.Context, addresses []string, connOptio
 				if azureConnection.Err() != nil {
 					Error("connection closed unexpectedly", "error", azureConnection.Err())
 					a.maybeReconnect()
+
 					return
 				}
 				Debug("connection closed successfully")
@@ -220,7 +222,6 @@ func (a *AmqpConnection) open(ctx context.Context, addresses []string, connOptio
 		return err
 	}
 
-	a.lifeCycle.SetState(&StateOpen{})
 	return nil
 }
 func (a *AmqpConnection) maybeReconnect() {
@@ -232,6 +233,7 @@ func (a *AmqpConnection) maybeReconnect() {
 	a.lifeCycle.SetState(&StateReconnecting{})
 	numberOfAttempts := 1
 	waitTime := a.amqpConnOptions.RecoveryConfiguration.BackOffReconnectInterval
+	reconnected := false
 	for numberOfAttempts <= a.amqpConnOptions.RecoveryConfiguration.MaxReconnectAttempts {
 		///wait for before reconnecting
 		Info("Waiting before reconnecting", "in", waitTime, "attempt", numberOfAttempts)
@@ -247,25 +249,29 @@ func (a *AmqpConnection) maybeReconnect() {
 			waitTime = waitTime * 2
 			Error("Failed to connection. ", "id", a.Id(), "error", err)
 		} else {
+			reconnected = true
 			break
 		}
 	}
 
-	var fails int32
-	Info("Reconnected successfully, restarting publishers and consumers")
-	a.entitiesTracker.publishers.Range(func(key, value any) bool {
-		publisher := value.(*Publisher)
-		// try to createSender
-		ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
-		err := publisher.createSender(ctx)
-		if err != nil {
-			atomic.AddInt32(&fails, 1)
-			Error("Failed to createSender publisher", "ID", publisher.Id(), "error", err)
-		}
-		cancel()
-		return true
-	})
-	Info("Restarted publishers", "number of fails", fails)
+	if reconnected {
+		var fails int32
+		Info("Reconnected successfully, restarting publishers and consumers")
+		a.entitiesTracker.publishers.Range(func(key, value any) bool {
+			publisher := value.(*Publisher)
+			// try to createSender
+			ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+			err := publisher.createSender(ctx)
+			if err != nil {
+				atomic.AddInt32(&fails, 1)
+				Error("Failed to createSender publisher", "ID", publisher.Id(), "error", err)
+			}
+			cancel()
+			return true
+		})
+		Info("Restarted publishers", "number of fails", fails)
+		a.lifeCycle.SetState(&StateOpen{})
+	}
 
 }
 

pkg/rabbitmq_amqp/amqp_connection_recovery_test.go

@@ -5,16 +5,27 @@ import (
 	"github.com/Azure/go-amqp"
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
-	test_helper "github.com/rabbitmq/rabbitmq-amqp-go-client/pkg/test-helper"
+	testhelper "github.com/rabbitmq/rabbitmq-amqp-go-client/pkg/test-helper"
 	"time"
 )
 
 var _ = Describe("Recovery connection test", func() {
-	It("connection should reconnect if dropped by via REST API", func() {
-		containerID := "connection should reconnect if dropped by via REST API"
+	It("connection should reconnect producers and consumers if dropped by via REST API", func() {
+		/*
+			The test is a bit complex since it requires to drop the connection by REST API
+			Then wait for the connection to be reconnected.
+			The scope of the test is to verify that the connection is reconnected and the
+			producers and consumers are able to send and receive messages.
+			It is more like an integration test.
+			This kind of the tests requires time in terms of execution it has to wait for the
+			connection to be reconnected, so to speed up the test I aggregated the tests in one.
+		*/
+
+		name := "connection should reconnect producers and consumers if dropped by via REST API"
 		connection, err := Dial(context.Background(), []string{"amqp://"}, &AmqpConnOptions{
 			SASLType:    amqp.SASLTypeAnonymous(),
-			ContainerID: containerID,
+			ContainerID: name,
+			// reduced the reconnect interval to speed up the test
 			RecoveryConfiguration: &RecoveryConfiguration{
 				ActiveRecovery:           true,
 				BackOffReconnectInterval: 2 * time.Second,
@@ -25,19 +36,77 @@ var _ = Describe("Recovery connection test", func() {
 		ch := make(chan *StateChanged, 1)
 		connection.NotifyStatusChange(ch)
 
+		qName := generateName(name)
+		queueInfo, err := connection.Management().DeclareQueue(context.Background(), &QuorumQueueSpecification{
+			Name: qName,
+		})
+		Expect(err).To(BeNil())
+		Expect(queueInfo).NotTo(BeNil())
+
+		publisher, err := connection.NewPublisher(context.Background(), &QueueAddress{
+			Queue: qName,
+		}, "test")
+
+		Expect(err).To(BeNil())
+		Expect(publisher).NotTo(BeNil())
+		for i := 0; i < 5; i++ {
+			publishResult, err := publisher.Publish(context.Background(), amqp.NewMessage([]byte("Hello")))
+			Expect(err).To(BeNil())
+			Expect(publishResult).NotTo(BeNil())
+			Expect(publishResult.Outcome).To(Equal(&amqp.StateAccepted{}))
+		}
+
 		Eventually(func() bool {
-			err := test_helper.DropConnectionContainerID(containerID)
+			err := testhelper.DropConnectionContainerID(name)
 			return err == nil
 		}).WithTimeout(5 * time.Second).WithPolling(400 * time.Millisecond).Should(BeTrue())
-		<-ch
-		time.Sleep(2 * time.Second)
+		st1 := <-ch
+		Expect(st1.From).To(Equal(&StateOpen{}))
+		Expect(st1.To).To(BeAssignableToTypeOf(&StateClosed{}))
+		///  Closed state should have an error
+		// Since it is forced closed by the REST API
+		err = st1.To.(*StateClosed).GetError()
+		Expect(err).NotTo(BeNil())
+		Expect(err.Error()).To(ContainSubstring("Connection forced"))
+
+		time.Sleep(1 * time.Second)
 		Eventually(func() bool {
-			conn, err := test_helper.GetConnectionByContainerID(containerID)
+			conn, err := testhelper.GetConnectionByContainerID(name)
 			return err == nil && conn != nil
 		}).WithTimeout(5 * time.Second).WithPolling(400 * time.Millisecond).Should(BeTrue())
-		<-ch
+		st2 := <-ch
+		Expect(st2.From).To(BeAssignableToTypeOf(&StateClosed{}))
+		Expect(st2.To).To(Equal(&StateReconnecting{}))
+
+		st3 := <-ch
+		Expect(st3.From).To(BeAssignableToTypeOf(&StateReconnecting{}))
+		Expect(st3.To).To(Equal(&StateOpen{}))
+
+		for i := 0; i < 5; i++ {
+			publishResult, err := publisher.Publish(context.Background(), amqp.NewMessage([]byte("Hello")))
+			Expect(err).To(BeNil())
+			Expect(publishResult).NotTo(BeNil())
+			Expect(publishResult.Outcome).To(Equal(&amqp.StateAccepted{}))
+		}
+
+		time.Sleep(500 * time.Millisecond)
+		purged, err := connection.Management().PurgeQueue(context.Background(), qName)
+		Expect(err).To(BeNil())
+		Expect(purged).To(Equal(5 + 5))
+
+		Expect(connection.Management().DeleteQueue(context.Background(), qName)).To(BeNil())
+
 		err = connection.Close(context.Background())
-		<-ch
+		Expect(err).To(BeNil())
+		st4 := <-ch
+		Expect(st4.From).To(Equal(&StateOpen{}))
+		Expect(st4.To).To(BeAssignableToTypeOf(&StateClosed{}))
+		err = st4.To.(*StateClosed).GetError()
+		// the flow status should be:
+		// from open to closed (with error)
+		// from closed to reconnecting
+		// from reconnecting to open
+		// from open to closed (without error)
 		Expect(err).To(BeNil())
 	})
 })

pkg/rabbitmq_amqp/amqp_management.go

@@ -13,6 +13,10 @@ import (
 var ErrPreconditionFailed = errors.New("precondition Failed")
 var ErrDoesNotExist = errors.New("does not exist")
 
+/*
+AmqpManagement is the interface to the RabbitMQ /management endpoint
+The management interface is used to declare/delete exchanges, queues, and bindings
+*/
 type AmqpManagement struct {
 	session   *amqp.Session
 	sender    *amqp.Sender
@@ -83,6 +87,11 @@ func (a *AmqpManagement) Close(ctx context.Context) error {
 	return err
 }
 
+/*
+Request sends a request to the /management endpoint.
+It is a generic method that can be used to send any request to the management endpoint.
+In most of the cases you don't need to use this method directly, instead use the standard methods
+*/
 func (a *AmqpManagement) Request(ctx context.Context, body any, path string, method string,
 	expectedResponseCodes []int) (map[string]any, error) {
 	return a.request(ctx, uuid.New().String(), body, path, method, expectedResponseCodes)

pkg/rabbitmq_amqp/entities.go

@@ -20,6 +20,9 @@ func (e QueueType) String() string {
 	return string(e.Type)
 }
 
+/*
+QueueSpecification represents the specification of a queue
+*/
 type QueueSpecification interface {
 	name() string
 	isAutoDelete() bool
@@ -28,8 +31,6 @@ type QueueSpecification interface {
 	buildArguments() map[string]any
 }
 
-// QuorumQueueSpecification represents the specification of the quorum queue
-
 type OverflowStrategy interface {
 	overflowStrategy() string
 }
@@ -73,6 +74,10 @@ func (r *ClientLocalLeaderLocator) leaderLocator() string {
 	return "client-local"
 }
 
+/*
+QuorumQueueSpecification represents the specification of the quorum queue
+*/
+
 type QuorumQueueSpecification struct {
 	Name                 string
 	AutoExpire           int64
@@ -154,7 +159,9 @@ func (q *QuorumQueueSpecification) buildArguments() map[string]any {
 	return result
 }
 
-// ClassicQueueSpecification represents the specification of the classic queue
+/*
+ClassicQueueSpecification represents the specification of the classic queue
+*/
 type ClassicQueueSpecification struct {
 	Name                 string
 	IsAutoDelete         bool
@@ -235,6 +242,11 @@ func (q *ClassicQueueSpecification) buildArguments() map[string]any {
 	return result
 }
 
+/*
+AutoGeneratedQueueSpecification represents the specification of the auto-generated queue.
+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 {
 	IsAutoDelete   bool
 	IsExclusive    bool

pkg/rabbitmq_amqp/life_cycle.go

@@ -65,13 +65,13 @@ func statusToString(status LifeCycleState) string {
 }
 
 type StateChanged struct {
-	From  LifeCycleState
-	To    LifeCycleState
-	Error error
+	From LifeCycleState
+	To   LifeCycleState
 }
 
 func (s StateChanged) String() string {
-	return fmt.Sprintf("From: %s, To: %s, Error: %v", statusToString(s.From), statusToString(s.To), s.Error)
+
+	return fmt.Sprintf("From: %s, To: %s", statusToString(s.From), statusToString(s.To))
 }
 
 type LifeCycle struct {
@@ -107,16 +107,8 @@ func (l *LifeCycle) SetState(value LifeCycleState) {
 		return
 	}
 
-	var stateError error
-	switch value.(type) {
-	case *StateClosed:
-		stateError = value.(*StateClosed).GetError()
-
-	}
-
 	l.chStatusChanged <- &StateChanged{
-		From:  oldState,
-		To:    value,
-		Error: stateError,
+		From: oldState,
+		To:   value,
 	}
 }