Improve support to AWS SQS/SNS

src/Paramore.Brighter.MessagingGateway.AWSSQS/AWSMessagingGateway.cs

@@ -1,4 +1,5 @@
 #region Licence
+
 /* The MIT License (MIT)
 Copyright © 2022 Ian Cooper <[email protected]>
 
@@ -25,6 +26,7 @@ THE SOFTWARE. */
 using System;
 using System.Collections.Generic;
 using System.Threading.Tasks;
+using Amazon.Runtime.Internal.Transform;
 using Amazon.SimpleNotificationService;
 using Amazon.SimpleNotificationService.Model;
 using Microsoft.Extensions.Logging;
@@ -46,43 +48,59 @@ public AWSMessagingGateway(AWSMessagingGatewayConnection awsConnection)
             _awsClientFactory = new AWSClientFactory(awsConnection);
         }
 
-        protected async Task<string> EnsureTopicAsync(RoutingKey topic, SnsAttributes attributes, TopicFindBy topicFindBy, OnMissingChannel makeTopic)
+        protected async Task<string> EnsureTopicAsync(RoutingKey topic, SnsAttributes attributes,
+            TopicFindBy topicFindBy, OnMissingChannel makeTopic, SnsSqsType snsSqsType, bool deduplication)
         {
             //on validate or assume, turn a routing key into a topicARN
-            if ((makeTopic == OnMissingChannel.Assume) || (makeTopic == OnMissingChannel.Validate)) 
-                await ValidateTopicAsync(topic, topicFindBy, makeTopic);
-            else if (makeTopic == OnMissingChannel.Create) CreateTopic(topic, attributes);
+            if ((makeTopic == OnMissingChannel.Assume) || (makeTopic == OnMissingChannel.Validate))
+                await ValidateTopicAsync(topic, topicFindBy, makeTopic, snsSqsType);
+            else if (makeTopic == OnMissingChannel.Create) CreateTopic(topic, attributes, snsSqsType, deduplication);
             return ChannelTopicArn;
         }
 
-        private void CreateTopic(RoutingKey topicName, SnsAttributes snsAttributes)
+        private void CreateTopic(RoutingKey topicName, SnsAttributes snsAttributes, SnsSqsType snsSqsType, bool deduplication)
         {
             using var snsClient = _awsClientFactory.CreateSnsClient();
             var attributes = new Dictionary<string, string>();
             if (snsAttributes != null)
             {
-                if (!string.IsNullOrEmpty(snsAttributes.DeliveryPolicy)) attributes.Add("DeliveryPolicy", snsAttributes.DeliveryPolicy);
+                if (!string.IsNullOrEmpty(snsAttributes.DeliveryPolicy))
+                    attributes.Add("DeliveryPolicy", snsAttributes.DeliveryPolicy);
                 if (!string.IsNullOrEmpty(snsAttributes.Policy)) attributes.Add("Policy", snsAttributes.Policy);
             }
 
-            var createTopicRequest = new CreateTopicRequest(topicName)
+            string name = topicName;
+            if (snsSqsType == SnsSqsType.Fifo)
             {
-                Attributes = attributes,
-                Tags = new List<Tag> {new Tag {Key = "Source", Value = "Brighter"}}
-            };
+                name += ".fifo";
 
+                attributes.Add("FifoTopic", "true");
+                if (deduplication)
+                {
+                    attributes.Add("ContentBasedDeduplication", "true");
+                }
+            }
+
+            var createTopicRequest = new CreateTopicRequest(name)
+            {
+                Attributes = attributes, Tags = new List<Tag> { new Tag { Key = "Source", Value = "Brighter" } }
+            };
+
+
             //create topic is idempotent, so safe to call even if topic already exists
             var createTopic = snsClient.CreateTopicAsync(createTopicRequest).Result;
-                
+
             if (!string.IsNullOrEmpty(createTopic.TopicArn))
                 ChannelTopicArn = createTopic.TopicArn;
             else
-                throw new InvalidOperationException($"Could not create Topic topic: {topicName} on {_awsConnection.Region}");
+                throw new InvalidOperationException(
+                    $"Could not create Topic topic: {name} on {_awsConnection.Region}");
         }
 
-        private async Task ValidateTopicAsync(RoutingKey topic, TopicFindBy findTopicBy, OnMissingChannel onMissingChannel)
+        private async Task ValidateTopicAsync(RoutingKey topic, TopicFindBy findTopicBy,
+            OnMissingChannel onMissingChannel, SnsSqsType snsSqsType)
         {
-            IValidateTopic topicValidationStrategy = GetTopicValidationStrategy(findTopicBy);
+            IValidateTopic topicValidationStrategy = GetTopicValidationStrategy(findTopicBy, snsSqsType);
             (bool exists, string topicArn) = await topicValidationStrategy.ValidateAsync(topic);
             if (exists)
                 ChannelTopicArn = topicArn;
@@ -91,16 +109,19 @@ private async Task ValidateTopicAsync(RoutingKey topic, TopicFindBy findTopicBy,
                     $"Topic validation error: could not find topic {topic}. Did you want Brighter to create infrastructure?");
         }
 
-        private IValidateTopic GetTopicValidationStrategy(TopicFindBy findTopicBy)
+        private IValidateTopic GetTopicValidationStrategy(TopicFindBy findTopicBy, SnsSqsType type)
         {
             switch (findTopicBy)
             {
                 case TopicFindBy.Arn:
-                    return new ValidateTopicByArn(_awsConnection.Credentials, _awsConnection.Region, _awsConnection.ClientConfigAction);
+                    return new ValidateTopicByArn(_awsConnection.Credentials, _awsConnection.Region,
+                        _awsConnection.ClientConfigAction);
                 case TopicFindBy.Convention:
-                    return new ValidateTopicByArnConvention(_awsConnection.Credentials, _awsConnection.Region, _awsConnection.ClientConfigAction);
+                    return new ValidateTopicByArnConvention(_awsConnection.Credentials, _awsConnection.Region,
+                        _awsConnection.ClientConfigAction, type);
                 case TopicFindBy.Name:
-                    return new ValidateTopicByName(_awsConnection.Credentials, _awsConnection.Region, _awsConnection.ClientConfigAction);
+                    return new ValidateTopicByName(_awsConnection.Credentials, _awsConnection.Region,
+                        _awsConnection.ClientConfigAction, type);
                 default:
                     throw new ConfigurationException("Unknown TopicFindBy used to determine how to read RoutingKey");
             }

src/Paramore.Brighter.MessagingGateway.AWSSQS/ChannelFactory.cs

@@ -81,7 +81,7 @@ public IAmAChannel CreateChannel(Subscription subscription)
                 SqsSubscription sqsSubscription = subscription as SqsSubscription;
                 _subscription = sqsSubscription ?? throw new ConfigurationException("We expect an SqsSubscription or SqsSubscription<T> as a parameter");
 
-                EnsureTopicAsync(_subscription.RoutingKey, _subscription.SnsAttributes, _subscription.FindTopicBy, _subscription.MakeChannels).Wait();
+                EnsureTopicAsync(_subscription.RoutingKey, _subscription.SnsAttributes, _subscription.FindTopicBy, _subscription.MakeChannels, _subscription.SqsType, false).Wait();
                 EnsureQueue();
 
                 return new Channel(
@@ -164,7 +164,19 @@ private void CreateQueue(AmazonSQSClient sqsClient)
                     }
                 }
 
-                var request = new CreateQueueRequest(_subscription.ChannelName.Value)
+
+                var queueName = _subscription.ChannelName.Value;
+                if (_subscription.SqsType == SnsSqsType.Fifo)
+                {
+                    if (!queueName.EndsWith(".fifo"))
+                    {
+                        queueName += ".fifo";
+                    }
+
+                    attributes.Add(QueueAttributeName.FifoQueue, "true");
+                }
+
+                var request = new CreateQueueRequest(queueName)
                 {
                     Attributes = attributes,
                     Tags = tags

src/Paramore.Brighter.MessagingGateway.AWSSQS/HeaderNames.cs

@@ -34,5 +34,7 @@ public static class HeaderNames
         public static readonly string Timestamp = "timestamp";
         public static readonly string ReplyTo = "reply-to";
         public static string Bag = "bag"; 
+        public const string MessageGroupId = "MessageGroupId";
+        public const string DeduplicationId = "MessageDeduplicationId";
     }
 }

src/Paramore.Brighter.MessagingGateway.AWSSQS/SnsPublication.cs

@@ -1,4 +1,5 @@
 #region Licence
+
 /* The MIT License (MIT)
 Copyright © 2022 Ian Cooper <[email protected]>
 
@@ -19,6 +20,7 @@ FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 THE SOFTWARE. */
+
 #endregion
 
 namespace Paramore.Brighter.MessagingGateway.AWSSQS
@@ -31,9 +33,8 @@ public class SnsPublication : Publication
         /// TopicFindBy.Convention -> The routing key is a name, but use convention to make an Arn for this account
         /// TopicFindBy.Name -> Treat the routing key as a name & use ListTopics to find it (rate limited 30/s)
         /// </summary>
-
         public TopicFindBy FindTopicBy { get; set; } = TopicFindBy.Convention;
-        
+
         /// <summary>
         /// The attributes of the topic. If TopicARNs is set we will always assume that we do not
         /// need to create or validate the SNS Topic
@@ -45,5 +46,42 @@ public class SnsPublication : Publication
         /// as we use the topic from the header to dispatch to  an Arn.
         /// </summary>
         public string TopicArn { get; set; }
+
+        /// <summary>
+        /// The AWS SQS type.
+        /// </summary>
+        public SnsSqsType SnsType { get; set; } = SnsSqsType.Standard;
+
+        /// <summary>
+        /// Amazon SNS FIFO topics and Amazon SQS FIFO queues support message deduplication, which provides
+        /// exactly-once message delivery and processing as long as the following conditions are met:
+        /// <list type="bullet">
+        ///     <item>
+        ///         <description>
+        ///             The subscribed Amazon SQS FIFO queue exists and has permissions that allow the
+        ///             AmazonSNS service principal to deliver messages to the queue.
+        ///         </description>
+        ///      </item>
+        ///     <item>
+        ///         <description>
+        ///             The Amazon SQS FIFO queue consumer processes the message and deletes it from the
+        ///             queue before the visibility timeout expires.
+        ///         </description>
+        ///      </item>
+        ///     <item>
+        ///         <description>
+        ///             The Amazon SNS subscription topic has no message filtering. When you configure
+        ///             message filtering, Amazon SNS FIFO topics support at-most-once delivery, as messages
+        ///             can be filtered out based on your subscription filter policies.
+        ///         </description>
+        ///      </item>
+        ///     <item>
+        ///         <description>
+        ///             There are no network disruptions that prevent acknowledgment of the message delivery
+        ///         </description>
+        ///      </item>
+        /// </list>
+        /// </summary>
+        public bool Deduplication { get; set; }
     }
 }

src/Paramore.Brighter.MessagingGateway.AWSSQS/SnsSqsType.cs

@@ -0,0 +1,42 @@
+namespace Paramore.Brighter.MessagingGateway.AWSSQS;
+
+/// <summary>
+/// AWS offer two types of SQS.
+/// For more information see
+/// https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-queue-types.html
+/// </summary>
+public enum SnsSqsType
+{
+    /// <summary>
+    /// Standard queues support a very high, nearly unlimited number of API calls per second per
+    /// action (SendMessage, ReceiveMessage, or DeleteMessage). This high throughput makes them
+    /// ideal for use cases that require processing large volumes of messages quickly, such as
+    /// real-time data streaming or large-scale applications. While standard queues scale
+    /// automatically with demand, it's essential to monitor usage patterns to ensure optimal
+    /// performance, especially in regions with higher workloads.
+    /// </summary>
+    Standard,
+
+    /// <summary>
+    /// FIFO (First-In-First-Out) queues have all the capabilities of the standard queues,
+    /// but are designed to enhance messaging between applications when the order of operations
+    /// and events is critical, or where duplicates can't be tolerated.
+    /// The most important features of FIFO queues are FIFO (First-In-First-Out) delivery and
+    /// exactly-once processing:
+    /// <list type="bullet">
+    ///     <item>
+    ///         <description>
+    ///             The order in which messages are sent and received is strictly preserved and
+    ///             a message is delivered once and remains unavailable until a consumer processes
+    ///             and deletes it.
+    ///         </description>
+    ///      </item>
+    ///     <item>
+    ///         <description>
+    ///             Duplicates aren't introduced into the queue.
+    ///         </description>
+    ///      </item>
+    /// </list>
+    /// </summary>
+    Fifo
+}

src/Paramore.Brighter.MessagingGateway.AWSSQS/SqsInlineMessageCreator.cs

@@ -1,4 +1,5 @@
 #region Licence
+
 /* The MIT License (MIT)
 Copyright © 2022 Ian Cooper <[email protected]>
 
@@ -19,6 +20,7 @@ FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 THE SOFTWARE. */
+
 #endregion
 
 using System;
@@ -31,7 +33,7 @@ namespace Paramore.Brighter.MessagingGateway.AWSSQS
 {
     internal class SqsInlineMessageCreator : SqsMessageCreatorBase, ISqsMessageCreator
     {
-        private static readonly ILogger s_logger= ApplicationLogging.CreateLogger<SqsInlineMessageCreator>();
+        private static readonly ILogger s_logger = ApplicationLogging.CreateLogger<SqsInlineMessageCreator>();
 
         private Dictionary<string, JsonElement> _messageAttributes = new Dictionary<string, JsonElement>();
 
@@ -64,7 +66,9 @@ public Message CreateMessage(Amazon.SQS.Model.Message sqsMessage)
                 replyTo = ReadReplyTo();
                 subject = ReadMessageSubject(jsonDocument);
                 receiptHandle = ReadReceiptHandle(sqsMessage);
-
+                var partitionKey = ReadPartitionKey();
+                var deduplicationId = ReadMessageDeduplicationId();
+
                 //TODO:CLOUD_EVENTS parse from headers
 
                 var messageHeader = new MessageHeader(
@@ -80,7 +84,9 @@ public Message CreateMessage(Amazon.SQS.Model.Message sqsMessage)
                     handledCount: handledCount.Result,
                     dataSchema: null,
                     subject: subject.Result,
-                    delayed: TimeSpan.Zero);
+                    delayed: TimeSpan.Zero,
+                    partitionKey: partitionKey.Result
+                );
 
                 message = new Message(messageHeader, ReadMessageBody(jsonDocument));
 
@@ -91,6 +97,11 @@ public Message CreateMessage(Amazon.SQS.Model.Message sqsMessage)
                     message.Header.Bag.Add(key, bag[key]);
                 }
 
+                if (deduplicationId.Success)
+                {
+                    message.Header.Bag.Add("DeduplicationId", deduplicationId);
+                }
+
                 if (receiptHandle.Success)
                     message.Header.Bag.Add("ReceiptHandle", sqsMessage.ReceiptHandle);
             }
@@ -99,8 +110,8 @@ public Message CreateMessage(Amazon.SQS.Model.Message sqsMessage)
                 s_logger.LogWarning(e, "Failed to create message from Aws Sqs message");
                 message = FailureMessage(topic, messageId);
             }
-            
-            
+
+
             return message;
         }
 
@@ -149,7 +160,6 @@ private Dictionary<string, object> ReadMessageBag()
                 }
                 catch (Exception)
                 {
-
                 }
             }
 
@@ -272,5 +282,29 @@ private static MessageBody ReadMessageBody(JsonDocument jsonDocument)
 
             return new MessageBody(string.Empty);
         }
+
+        private HeaderResult<string> ReadPartitionKey()
+        {
+            if (_messageAttributes.TryGetValue(HeaderNames.MessageGroupId, out var value))
+            {
+                //we have an arn, and we want the topic
+                var messageGroupId = value.GetValueInString();
+                return new HeaderResult<string>(messageGroupId, true);
+            }
+
+            return new HeaderResult<string>(string.Empty, true);
+        }
+
+        private HeaderResult<string> ReadMessageDeduplicationId()
+        {
+            if (_messageAttributes.TryGetValue(HeaderNames.DeduplicationId, out var value))
+            {
+                //we have an arn, and we want the topic
+                var deduplicationId = value.GetValueInString();
+                return new HeaderResult<string>(deduplicationId, true);
+            }
+
+            return new HeaderResult<string>(string.Empty, true);
+        }
     }
 }