add consumer recovery

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

Author: Gsantomaggio

docs/examples/getting_started/main.go

@@ -86,16 +86,16 @@ func main() {
 			deliveryContext, err := consumer.Receive(ctx)
 			if errors.Is(err, context.Canceled) {
 				// The consumer was closed correctly
-				rabbitmq_amqp.Info("[Consumer]", "consumer closed. Context", err)
+				rabbitmq_amqp.Info("[NewConsumer]", "consumer closed. Context", err)
 				return
 			}
 			if err != nil {
 				// An error occurred receiving the message
-				rabbitmq_amqp.Error("[Consumer]", "Error receiving message", err)
+				rabbitmq_amqp.Error("[NewConsumer]", "Error receiving message", err)
 				return
 			}
 
-			rabbitmq_amqp.Info("[Consumer]", "Received message",
+			rabbitmq_amqp.Info("[NewConsumer]", "Received message",
 				fmt.Sprintf("%s", deliveryContext.Message().Data))
 
 			err = deliveryContext.Accept(context.Background())
@@ -115,7 +115,7 @@ func main() {
 		return
 	}
 
-	for i := 0; i < 10_000; i++ {
+	for i := 0; i < 1_000; i++ {
 		// Publish a message to the exchange
 		publishResult, err := publisher.Publish(context.Background(), amqp.NewMessage([]byte("Hello, World!"+fmt.Sprintf("%d", i))))
 		if err != nil {
@@ -125,16 +125,16 @@ func main() {
 		}
 		switch publishResult.Outcome.(type) {
 		case *amqp.StateAccepted:
-			rabbitmq_amqp.Info("[Publisher]", "Message accepted", publishResult.Message.Data[0])
+			rabbitmq_amqp.Info("[NewPublisher]", "Message accepted", publishResult.Message.Data[0])
 			break
 		case *amqp.StateReleased:
-			rabbitmq_amqp.Warn("[Publisher]", "Message was not routed", publishResult.Message.Data[0])
+			rabbitmq_amqp.Warn("[NewPublisher]", "Message was not routed", publishResult.Message.Data[0])
 			break
 		case *amqp.StateRejected:
-			rabbitmq_amqp.Warn("[Publisher]", "Message rejected", publishResult.Message.Data[0])
+			rabbitmq_amqp.Warn("[NewPublisher]", "Message rejected", publishResult.Message.Data[0])
 			stateType := publishResult.Outcome.(*amqp.StateRejected)
 			if stateType.Error != nil {
-				rabbitmq_amqp.Warn("[Publisher]", "Message rejected with error: %v", stateType.Error)
+				rabbitmq_amqp.Warn("[NewPublisher]", "Message rejected with error: %v", stateType.Error)
 			}
 			break
 		default:
@@ -153,13 +153,13 @@ func main() {
 	//Close the consumer
 	err = consumer.Close(context.Background())
 	if err != nil {
-		rabbitmq_amqp.Error("[Consumer]", err)
+		rabbitmq_amqp.Error("[NewConsumer]", err)
 		return
 	}
 	// Close the publisher
 	err = publisher.Close(context.Background())
 	if err != nil {
-		rabbitmq_amqp.Error("[Publisher]", err)
+		rabbitmq_amqp.Error("[NewPublisher]", err)
 		return
 	}
 

docs/examples/reliable/reliable.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"github.com/Azure/go-amqp"
 	"github.com/rabbitmq/rabbitmq-amqp-go-client/pkg/rabbitmq_amqp"
@@ -14,6 +15,7 @@ func main() {
 	var stateAccepted int32
 	var stateReleased int32
 	var stateRejected int32
+
 	var received int32
 	var failed int32
 
@@ -40,6 +42,12 @@ func main() {
 	amqpConnection, err := rabbitmq_amqp.Dial(context.Background(), []string{"amqp://"}, &rabbitmq_amqp.AmqpConnOptions{
 		SASLType:    amqp.SASLTypeAnonymous(),
 		ContainerID: "reliable-amqp10-go",
+		RecoveryConfiguration: &rabbitmq_amqp.RecoveryConfiguration{
+			ActiveRecovery:           true,
+			BackOffReconnectInterval: 2 * time.Second, // we reduce the reconnect interval to speed up the test. The default is 5 seconds
+			// In production, you should avoid BackOffReconnectInterval with low values since it can cause a high number of reconnection attempts
+			MaxReconnectAttempts: 5,
+		},
 	})
 	if err != nil {
 		rabbitmq_amqp.Error("Error opening connection", err)
@@ -62,40 +70,45 @@ func main() {
 		return
 	}
 
-	//consumer, err := amqpConnection.NewConsumer(context.Background(), &rabbitmq_amqp.QueueAddress{
-	//	Queue: queueName,
-	//}, "reliable-consumer")
-	//if err != nil {
-	//	rabbitmq_amqp.Error("Error creating consumer", err)
-	//	return
-	//}
-
-	//consumerContext, cancel := context.WithCancel(context.Background())
-
-	//// Consume messages from the queue
-	//go func(ctx context.Context) {
-	//	for {
-	//		deliveryContext, err := consumer.Receive(ctx)
-	//		if errors.Is(err, context.Canceled) {
-	//			// The consumer was closed correctly
-	//			return
-	//		}
-	//		if err != nil {
-	//			// An error occurred receiving the message
-	//			rabbitmq_amqp.Error("[Consumer]", "Error receiving message", err)
-	//			return
-	//		}
-	//
-	//		rabbitmq_amqp.Info("[Consumer]", "Received message",
-	//			fmt.Sprintf("%s", deliveryContext.Message().Data))
-	//
-	//		err = deliveryContext.Accept(context.Background())
-	//		if err != nil {
-	//			rabbitmq_amqp.Error("Error accepting message", err)
-	//			return
-	//		}
-	//	}
-	//}(consumerContext)
+	consumer, err := amqpConnection.NewConsumer(context.Background(), &rabbitmq_amqp.QueueAddress{
+		Queue: queueName,
+	}, "reliable-consumer")
+	if err != nil {
+		rabbitmq_amqp.Error("Error creating consumer", err)
+		return
+	}
+
+	consumerContext, cancel := context.WithCancel(context.Background())
+
+	// Consume messages from the queue
+	go func(ctx context.Context) {
+		for {
+			deliveryContext, err := consumer.Receive(ctx)
+			if errors.Is(err, context.Canceled) {
+				// The consumer was closed correctly
+				return
+			}
+			if err != nil {
+				// An error occurred receiving the message
+				rabbitmq_amqp.Error("[NewConsumer]", "Error receiving message, retry in 2_500 ms", err, "queue", queueName)
+				// here the consumer could be disconnected from the server due to a network error
+				// in this specific case, we just wait for 2_500 ms and try again (2 seconds is the reconnect interval we defined + random 500 random ms)
+				// while the connection is reestablished
+				// you can use the stateChanged channel to be notified when the connection is reestablished
+				time.Sleep(2500 * time.Millisecond)
+				continue
+			}
+
+			atomic.AddInt32(&received, 1)
+			err = deliveryContext.Accept(context.Background())
+			if err != nil {
+				// same here the delivery could not be accepted due to a network error
+				// we wait for 2_500 ms and try again
+				time.Sleep(2500 * time.Millisecond)
+				continue
+			}
+		}
+	}(consumerContext)
 
 	publisher, err := amqpConnection.NewPublisher(context.Background(), &rabbitmq_amqp.QueueAddress{
 		Queue: queueName,
@@ -106,12 +119,16 @@ func main() {
 	}
 
 	for i := 0; i < 1_000_000; i++ {
-		// Publish a message to the exchange
 		publishResult, err := publisher.Publish(context.Background(), amqp.NewMessage([]byte("Hello, World!"+fmt.Sprintf("%d", i))))
 		if err != nil {
 			rabbitmq_amqp.Error("Error publishing message", "error", err)
+			// here you need to deal with the error. You can store the message in a local in memory/persistent storage
+			// then retry to send the message as soon as the connection is reestablished
+			// in this specific case, we just wait for 2_500 ms and try again (2 seconds is the reconnect interval we defined + random 500 random ms)
+			// you can use the stateChanged channel to be notified when the connection is reestablished
+			// and use some signal to reactivate the message sending
 			atomic.AddInt32(&failed, 1)
-			time.Sleep(1 * time.Second)
+			time.Sleep(2500 * time.Millisecond)
 			continue
 		}
 		switch publishResult.Outcome.(type) {
@@ -136,17 +153,17 @@ func main() {
 	var input string
 	_, _ = fmt.Scanln(&input)
 
-	//cancel()
+	cancel()
 	//Close the consumer
-	//err = consumer.Close(context.Background())
-	//if err != nil {
-	//	rabbitmq_amqp.Error("[Consumer]", err)
-	//	return
-	//}
+	err = consumer.Close(context.Background())
+	if err != nil {
+		rabbitmq_amqp.Error("[NewConsumer]", err)
+		return
+	}
 	// Close the publisher
 	err = publisher.Close(context.Background())
 	if err != nil {
-		rabbitmq_amqp.Error("[Publisher]", err)
+		rabbitmq_amqp.Error("[NewPublisher]", err)
 		return
 	}
 

pkg/rabbitmq_amqp/amqp_connection.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"github.com/Azure/go-amqp"
 	"github.com/google/uuid"
+	"math/rand"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -93,14 +94,7 @@ func (a *AmqpConnection) NewConsumer(ctx context.Context, destination *QueueAddr
 	}
 	err = validateAddress(destinationAdd)
 
-	if err != nil {
-		return nil, err
-	}
-	receiver, err := a.session.NewReceiver(ctx, destinationAdd, createReceiverLinkOptions(destinationAdd, linkName, AtLeastOnce))
-	if err != nil {
-		return nil, err
-	}
-	return newConsumer(receiver), nil
+	return newConsumer(ctx, a, destinationAdd, linkName)
 }
 
 // Dial connect to the AMQP 1.0 server using the provided connectionSettings
@@ -121,6 +115,16 @@ func Dial(ctx context.Context, addresses []string, connOptions *AmqpConnOptions,
 		connOptions.RecoveryConfiguration = NewRecoveryConfiguration()
 	}
 
+	// validate the RecoveryConfiguration options
+	if connOptions.RecoveryConfiguration.MaxReconnectAttempts <= 0 && connOptions.RecoveryConfiguration.ActiveRecovery {
+		return nil, fmt.Errorf("MaxReconnectAttempts should be greater than 0")
+	}
+	if connOptions.RecoveryConfiguration.BackOffReconnectInterval <= 1*time.Second && connOptions.RecoveryConfiguration.ActiveRecovery {
+		return nil, fmt.Errorf("BackOffReconnectInterval should be greater than 1 second")
+	}
+
+	// create the connection
+
 	conn := &AmqpConnection{
 		management:      NewAmqpManagement(),
 		lifeCycle:       NewLifeCycle(),
@@ -236,6 +240,10 @@ func (a *AmqpConnection) maybeReconnect() {
 	reconnected := false
 	for numberOfAttempts <= a.amqpConnOptions.RecoveryConfiguration.MaxReconnectAttempts {
 		///wait for before reconnecting
+		// add some random milliseconds to the wait time to avoid thundering herd
+		// the random time is between 0 and 500 milliseconds
+		waitTime = waitTime + time.Duration(rand.Intn(500))*time.Millisecond
+
 		Info("Waiting before reconnecting", "in", waitTime, "attempt", numberOfAttempts)
 		time.Sleep(waitTime)
 		// context with timeout
@@ -270,6 +278,21 @@ func (a *AmqpConnection) maybeReconnect() {
 			return true
 		})
 		Info("Restarted publishers", "number of fails", fails)
+		fails = 0
+		a.entitiesTracker.consumers.Range(func(key, value any) bool {
+			consumer := value.(*Consumer)
+			// try to createSender
+			ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+			err := consumer.createReceiver(ctx)
+			if err != nil {
+				atomic.AddInt32(&fails, 1)
+				Error("Failed to createReceiver consumer", "ID", consumer.Id(), "error", err)
+			}
+			cancel()
+			return true
+		})
+		Info("Restarted consumers", "number of fails", fails)
+
 		a.lifeCycle.SetState(&StateOpen{})
 	}
 

pkg/rabbitmq_amqp/amqp_connection_recovery.go

@@ -16,12 +16,15 @@ type RecoveryConfiguration struct {
 		BackOffReconnectInterval The time to wait before trying to createSender after a connection is closed.
 		time will be increased exponentially with each attempt.
 		Default is 5 seconds, each attempt will double the time.
+		The minimum value is 1 second. Avoid setting a value low values since it can cause a high
+		number of reconnection attempts.
 	*/
 	BackOffReconnectInterval time.Duration
 
 	/*
 		MaxReconnectAttempts The maximum number of reconnection attempts.
 		Default is 5.
+		The minimum value is 1.
 	*/
 	MaxReconnectAttempts int
 }