add error

max ErrMaxReconnectAttemptsReached

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

Author: Gsantomaggio

docs/examples/reliable/reliable.go

@@ -16,18 +16,19 @@ func main() {
 	var stateAccepted int32
 	var stateReleased int32
 	var stateRejected int32
+	var isRunning bool
 
 	var received int32
 	var failed int32
 
 	startTime := time.Now()
+	isRunning = true
 	go func() {
-		for {
+		for isRunning {
 			time.Sleep(5 * time.Second)
 			total := stateAccepted + stateReleased + stateRejected
 			messagesPerSecond := float64(total) / time.Since(startTime).Seconds()
 			rmq.Info("[Stats]", "sent", total, "received", received, "failed", failed, "messagesPerSecond", messagesPerSecond)
-
 		}
 	}()
 
@@ -41,6 +42,16 @@ func main() {
 			switch statusChanged.To.(type) {
 			case *rmq.StateOpen:
 				signalBlock.Broadcast()
+			case *rmq.StateReconnecting:
+				rmq.Info("[connection]", "Reconnecting to the AMQP 1.0 server")
+			case *rmq.StateClosed:
+				StateClosed := statusChanged.To.(*rmq.StateClosed)
+				if errors.Is(StateClosed.GetError(), rmq.ErrMaxReconnectAttemptsReached) {
+					rmq.Error("[connection]", "Max reconnect attempts reached. Closing connection", StateClosed.GetError())
+					signalBlock.Broadcast()
+					isRunning = false
+				}
+
 			}
 		}
 	}(stateChanged)
@@ -130,6 +141,10 @@ func main() {
 		go func() {
 			defer wg.Done()
 			for i := 0; i < 500_000; i++ {
+				if !isRunning {
+					rmq.Info("[Publisher]", "Publisher is stopped simulation not running, queue", queueName)
+					return
+				}
 				publishResult, err := publisher.Publish(context.Background(), rmq.NewMessage([]byte("Hello, World!"+fmt.Sprintf("%d", i))))
 				if err != nil {
 					// here you need to deal with the error. You can store the message in a local in memory/persistent storage

pkg/rabbitmqamqp/amqp_binding.go

@@ -58,8 +58,8 @@ func (b *AMQPBinding) Bind(ctx context.Context) (string, error) {
 	kv[destination] = b.destinationName
 	kv["arguments"] = make(map[string]any)
 	_, err := b.management.Request(ctx, kv, path, commandPost, []int{responseCode204})
-	bindingPathWithExchangeQueueKey := bindingPathWithExchangeQueueKey(b.toQueue, b.sourceName, b.destinationName, b.bindingKey)
-	return bindingPathWithExchangeQueueKey, err
+	bindingPathWithExchangeQueueAndKey := bindingPathWithExchangeQueueKey(b.toQueue, b.sourceName, b.destinationName, b.bindingKey)
+	return bindingPathWithExchangeQueueAndKey, err
 }
 
 // Unbind removes a binding between an exchange and a queue or exchange

pkg/rabbitmqamqp/amqp_connection.go

@@ -350,6 +350,10 @@ func (a *AmqpConnection) maybeReconnect() {
 		Error("Reconnection attempt failed", "attempt", attempt, "error", err, "ID", a.Id())
 	}
 
+	// If we reach here, all attempts failed
+	Error("All reconnection attempts failed, closing connection", "ID", a.Id())
+	a.lifeCycle.SetState(&StateClosed{error: ErrMaxReconnectAttemptsReached})
+
 }
 
 // restartEntities attempts to restart all publishers and consumers after a reconnection

pkg/rabbitmqamqp/amqp_connection_recovery.go

@@ -1,10 +1,14 @@
 package rabbitmqamqp
 
 import (
+	"errors"
 	"sync"
 	"time"
 )
 
+// ErrMaxReconnectAttemptsReached typed error when the MaxReconnectAttempts is reached
+var ErrMaxReconnectAttemptsReached = errors.New("max reconnect attempts reached, connection will not be recovered")
+
 type RecoveryConfiguration struct {
 	/*
 		ActiveRecovery Define if the recovery is activated.