Added fatal checker configuration option in SQSMessagePollerOptions (#62

)

src/AWS.Messaging/Configuration/MessageBusBuilder.cs

@@ -123,7 +123,9 @@ public IMessageBusBuilder AddSQSPoller(string queueUrl, Action<SQSMessagePollerO
  VisibilityTimeout = sqsMessagePollerOptions.VisibilityTimeout,
  VisibilityTimeoutExtensionThreshold = sqsMessagePollerOptions.VisibilityTimeoutExtensionThreshold,
  VisibilityTimeoutExtensionHeartbeatInterval = sqsMessagePollerOptions.VisibilityTimeoutExtensionHeartbeatInterval,
- WaitTimeSeconds = sqsMessagePollerOptions.WaitTimeSeconds
+ WaitTimeSeconds = sqsMessagePollerOptions.WaitTimeSeconds,
+ IsSQSExceptionFatal = sqsMessagePollerOptions.IsSQSExceptionFatal
+
  };
 
  _messageConfiguration.MessagePollerConfigurations.Add(sqsMessagePollerConfiguration);

src/AWS.Messaging/Configuration/SQSMessagePollerConfiguration.cs

@@ -1,6 +1,7 @@
 // Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 // SPDX-License-Identifier: Apache-2.0
 
+using Amazon.SQS;
 using Amazon.SQS.Model;
 
 namespace AWS.Messaging.Configuration;
@@ -87,6 +88,11 @@ internal class SQSMessagePollerConfiguration : IMessagePollerConfiguration
  /// </remarks>
  public int WaitTimeSeconds { get; init; } = DEFAULT_WAIT_TIME_SECONDS;
 
+ /// <summary>
+ /// Function that indicates whether an <see cref="AmazonSQSException"/> should stop the poller or continue.
+ /// </summary>
+ public Func<AmazonSQSException, bool> IsSQSExceptionFatal { get; set; } = IsFatalException;
+
  /// <summary>
  /// Construct an instance of <see cref="SQSMessagePollerConfiguration" />
  /// </summary>
@@ -113,4 +119,22 @@ internal MessageManagerConfiguration ToMessageManagerConfiguration()
  VisibilityTimeoutExtensionHeartbeatInterval = VisibilityTimeoutExtensionHeartbeatInterval
  };
  }
+
+ /// <summary>
+ /// <see cref="AmazonSQSException"/> error codes that should be treated as fatal and stop the poller
+ /// </summary>
+ private static readonly HashSet<string> _fatalSQSErrorCodes = new HashSet<string>
+ {
+ "InvalidAddress", // Returned for an invalid queue URL
+ "AccessDenied" // Returned with insufficient IAM permissions to read from the configured queue
+ };
+
+ /// <summary>
+ /// Determines if a given SQS exception should be treated as fatal and rethrown to stop the poller
+ /// </summary>
+ /// <param name="sqsException">SQS Exception</param>
+ internal static bool IsFatalException(AmazonSQSException sqsException)
+ {
+ return sqsException is QueueDoesNotExistException ? true : _fatalSQSErrorCodes.Contains(sqsException.ErrorCode);
+ }
 }

src/AWS.Messaging/Configuration/SQSMessagePollerOptions.cs

@@ -1,6 +1,8 @@
 // Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 // SPDX-License-Identifier: Apache-2.0
 
+using Amazon.SQS;
+
 namespace AWS.Messaging.Configuration;
 
 /// <summary>
@@ -23,6 +25,11 @@ public class SQSMessagePollerOptions
  /// <inheritdoc cref="SQSMessagePollerConfiguration.VisibilityTimeoutExtensionHeartbeatInterval"/>
  public int VisibilityTimeoutExtensionHeartbeatInterval { get; set; } = SQSMessagePollerConfiguration.DEFAULT_VISIBILITY_TIMEOUT_EXTENSION_HEARTBEAT_INTERVAL;
 
+ /// <summary>
+ /// <inheritdoc cref="SQSMessagePollerConfiguration.IsFatalException(AmazonSQSException)" />
+ /// </summary>
+ public Func<AmazonSQSException, bool> IsSQSExceptionFatal { get; set; } = SQSMessagePollerConfiguration.IsFatalException;
+
  /// <summary>
  /// Validates that the options are valid against the message framework's and/or SQS limits
  /// </summary>

src/AWS.Messaging/SQS/SQSMessagePoller.cs

@@ -124,7 +124,7 @@ private async Task PollQueue(CancellationToken token)
 
  // Rethrow the exception to fail fast for invalid configuration, permissioning, etc.
  // TODO: explore a "cool down mode" for repeated exceptions
- if (IsSQSExceptionFatal(ex))
+ if (_configuration.IsSQSExceptionFatal(ex))
  {
  throw;
  }
@@ -213,7 +213,7 @@ public async Task DeleteMessagesAsync(IEnumerable<MessageEnvelope> messages, Can
  string.Join(", ", messages.Select(x => x.Id)), _configuration.SubscriberEndpoint);
 
  // Rethrow the exception to fail fast for invalid configuration, permissioning, etc.
- if (IsSQSExceptionFatal(ex))
+ if (_configuration.IsSQSExceptionFatal(ex))
  {
  throw;
  }
@@ -319,7 +319,7 @@ public async Task ExtendMessageVisibilityTimeoutAsync(IEnumerable<MessageEnvelop
  string.Join(", ", messages.Select(x => x.Id)), _configuration.SubscriberEndpoint);
 
  // Rethrow the exception to fail fast for invalid configuration, permissioning, etc.
- if (IsSQSExceptionFatal(amazonEx))
+ if (_configuration.IsSQSExceptionFatal(amazonEx))
  {
  throw amazonEx;
  }
@@ -338,22 +338,5 @@ public ValueTask ReportMessageFailureAsync(MessageEnvelope message, Cancellation
  {
  return ValueTask.CompletedTask;
  }
-
- /// <summary>
- /// <see cref="AmazonSQSException"/> error codes that should be treated as fatal and stop the poller
- /// </summary>
- private static readonly HashSet<string> _fatalSQSErrorCodes = new HashSet<string>
- {
- "InvalidAddress", // Returned for an invalid queue URL
- "AccessDenied" // Returned with insufficient IAM permissions to read from the configured queue
- };
-
- /// <summary>
- /// Determines if a given SQS exception should be treated as fatal and rethrown to stop the poller
- /// </summary>
- /// <param name="sqsException">SQS Exception</param>
- private bool IsSQSExceptionFatal(AmazonSQSException sqsException)
- {
- return _fatalSQSErrorCodes.Contains(sqsException.ErrorCode);
- }
+
 }