Only retry with retryable amqp errors for sender

Author: gavinfish

errors.go

@@ -3,8 +3,25 @@ package servicebus
 import (
 	"fmt"
 	"reflect"
+	"time"
 
 	"github.com/Azure/azure-amqp-common-go/v3/rpc"
+	"github.com/Azure/go-amqp"
+)
+
+// Error Conditions
+const (
+	// Service Bus Errors
+	ErrorServerBusy         amqp.ErrorCondition = "com.microsoft:server-busy"
+	ErrorTimeout            amqp.ErrorCondition = "com.microsoft:timeout"
+	ErrorOperationCancelled amqp.ErrorCondition = "com.microsoft:operation-cancelled"
+	ErrorContainerClose     amqp.ErrorCondition = "com.microsoft:container-close"
+)
+
+const (
+	AmqpRetryDefaultTimes    int           = 3
+	AmqpRetryDefaultDelay    time.Duration = time.Second
+	AmqpRetryBusyServerDelay time.Duration = 10 * time.Second
 )
 
 type (

sender.go

@@ -234,31 +234,10 @@ func (s *Sender) trySend(ctx context.Context, evt eventer) error {
 
 			switch err.(type) {
 			case *amqp.Error, *amqp.DetachError:
-				tab.For(ctx).Debug("recovering connection")
-				_, retryErr := common.Retry(10, 10*time.Second, func() (interface{}, error) {
-					ctx, sp := s.startProducerSpanFromContext(ctx, "sb.Sender.trySend.tryRecover")
-					defer sp.End()
-
-					err := s.Recover(ctx)
-					if err == nil {
-						tab.For(ctx).Debug("recovered connection")
-						return nil, nil
-					}
-
-					select {
-					case <-ctx.Done():
-						return nil, ctx.Err()
-					default:
-						return nil, common.Retryable(err.Error())
-					}
-				})
-
-				if retryErr != nil {
-					tab.For(ctx).Debug("sender recovering retried, but error was unrecoverable")
-					if err := s.Close(ctx); err != nil {
-						tab.For(ctx).Error(err)
-					}
-					return retryErr
+				err = s.handleAmqpError(ctx, err)
+				if err != nil {
+					tab.For(ctx).Error(err)
+					return err
 				}
 			default:
 				tab.For(ctx).Error(err)
@@ -268,6 +247,50 @@ func (s *Sender) trySend(ctx context.Context, evt eventer) error {
 	}
 }
 
+func (s *Sender) handleAmqpError(ctx context.Context, err error) error {
+	if amqpError, ok := err.(*amqp.Error); ok {
+		switch amqpError.Condition {
+		case ErrorServerBusy:
+			return s.retryRetryableAmqpError(ctx, AmqpRetryDefaultTimes, AmqpRetryBusyServerDelay)
+		case ErrorTimeout:
+			return s.retryRetryableAmqpError(ctx, AmqpRetryDefaultTimes, AmqpRetryDefaultDelay)
+		case ErrorOperationCancelled:
+			return s.retryRetryableAmqpError(ctx, AmqpRetryDefaultTimes, AmqpRetryDefaultDelay)
+		case ErrorContainerClose:
+			return s.retryRetryableAmqpError(ctx, AmqpRetryDefaultTimes, AmqpRetryDefaultDelay)
+		default:
+			return err
+		}
+	}
+	return s.retryRetryableAmqpError(ctx, AmqpRetryDefaultTimes, AmqpRetryDefaultDelay)
+}
+
+func (s *Sender) retryRetryableAmqpError(ctx context.Context, times int, delay time.Duration) error {
+	tab.For(ctx).Debug("recovering sender connection")
+	_, retryErr := common.Retry(times, delay, func() (interface{}, error) {
+		ctx, sp := s.startProducerSpanFromContext(ctx, "sb.Sender.trySend.tryRecover")
+		defer sp.End()
+
+		err := s.Recover(ctx)
+		if err == nil {
+			tab.For(ctx).Debug("recovered connection")
+			return nil, nil
+		}
+
+		select {
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		default:
+			return nil, common.Retryable(err.Error())
+		}
+	})
+	if retryErr != nil {
+		tab.For(ctx).Debug("sender recovering retried, but error was unrecoverable")
+		return retryErr
+	}
+	return nil
+}
+
 func (s *Sender) connClosedError(ctx context.Context) error {
 	name := "Sender"
 	if s.Name != "" {