GH-1294 Add support to FIFO

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 @@
 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 @@
 {
     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 @@
                 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 @@
                     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 @@
                     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 @@
                 s_logger.LogWarning(e, "Failed to create message from Aws Sqs message");
                 message = FailureMessage(topic, messageId);
             }
-            
-            
+
+
             return message;
         }
 
@@ -149,7 +160,6 @@
                 }
                 catch (Exception)
                 {
-
                 }
             }
 
@@ -272,5 +282,29 @@
 
             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);
+        }
     }
 }