[Host.AmazonSQS] Add SNS support

Signed-off-by: Tomasz Maruszak <maruszaktomasz@gmail.com>

Author: zarusz

src/SlimMessageBus.Host.AmazonSQS/ClientFactory/NullSnsClientProvider.cs

@@ -2,7 +2,8 @@
 
 internal class NullSnsClientProvider : ISnsClientProvider
 {
-    public AmazonSimpleNotificationServiceClient Client => null;
+    public AmazonSimpleNotificationServiceClient Client
+        => throw new ConfigurationMessageBusException("The connection to Amazon SNS has not been provided - check your bus configuration");
 
     public Task EnsureClientAuthenticated() => Task.CompletedTask;
 }

src/SlimMessageBus.Host.AmazonSQS/ClientFactory/NullSqsClientProvider.cs

@@ -2,7 +2,8 @@
 
 internal class NullSqsClientProvider : ISqsClientProvider
 {
-    public AmazonSQSClient Client => null;
+    public AmazonSQSClient Client
+        => throw new ConfigurationMessageBusException("The connection to Amazon SQS has not been provided - check your bus configuration");
 
     public Task EnsureClientAuthenticated() => Task.CompletedTask;
 }

src/SlimMessageBus.Host.AmazonSQS/Consumer/SqsBaseConsumer.cs

@@ -13,7 +13,7 @@ public abstract class SqsBaseConsumer : AbstractConsumer
 
     public SqsMessageBus MessageBus { get; }
     protected IMessageProcessor<Message> MessageProcessor { get; }
-    protected ISqsHeaderSerializer HeaderSerializer { get; }
+    protected ISqsHeaderSerializer<Amazon.SQS.Model.MessageAttributeValue> HeaderSerializer { get; }
 
     protected SqsBaseConsumer(
         SqsMessageBus messageBus,
@@ -30,7 +30,7 @@ protected SqsBaseConsumer(
         _clientProvider = clientProvider ?? throw new ArgumentNullException(nameof(clientProvider));
         MessageBus = messageBus;
         MessageProcessor = messageProcessor ?? throw new ArgumentNullException(nameof(messageProcessor));
-        HeaderSerializer = messageBus.HeaderSerializer;
+        HeaderSerializer = messageBus.SqsHeaderSerializer;
         T GetSingleValue<T>(Func<AbstractConsumerSettings, T> selector, string settingName, T defaultValue = default)
         {
             var set = consumerSettings.Select(x => selector(x)).Where(x => x is not null && !x.Equals(defaultValue)).ToHashSet();
@@ -43,7 +43,7 @@ T GetSingleValue<T>(Func<AbstractConsumerSettings, T> selector, string settingNa
 
         _maxMessages = GetSingleValue(x => x.GetOrDefault(SqsProperties.MaxMessages), nameof(SqsConsumerBuilderExtensions.MaxMessages)) ?? messageBus.ProviderSettings.MaxMessageCount;
         _visibilityTimeout = GetSingleValue(x => x.GetOrDefault(SqsProperties.VisibilityTimeout), nameof(SqsConsumerBuilderExtensions.VisibilityTimeout)) ?? 30;
-        _messageAttributeNames = new List<string>(GetSingleValue(x => x.GetOrDefault(SqsProperties.MessageAttributes), nameof(SqsConsumerBuilderExtensions.FetchMessageAttributes)) ?? ["All"]);
+        _messageAttributeNames = [.. GetSingleValue(x => x.GetOrDefault(SqsProperties.MessageAttributes), nameof(SqsConsumerBuilderExtensions.FetchMessageAttributes)) ?? ["All"]];
     }
 
     private async Task<IReadOnlyCollection<Message>> ReceiveMessagesByUrl(string queueUrl)
@@ -101,7 +101,7 @@ protected override async Task OnStop()
 
     protected async Task Run()
     {
-        var queueUrl = MessageBus.GetQueueUrlOrException(Path);
+        var (queueUrl, _) = MessageBus.GetUrlAndKindOrException(Path);
 
         var messagesToDelete = new List<Message>(_maxMessages);
 

src/SlimMessageBus.Host.AmazonSQS/Headers/DefaultSnsHeaderSerializer.cs

@@ -0,0 +1,32 @@
+namespace SlimMessageBus.Host.AmazonSQS;
+
+public class DefaultSnsHeaderSerializer(bool detectStringType = true) : ISqsHeaderSerializer<Amazon.SimpleNotificationService.Model.MessageAttributeValue>
+{
+    const string DataTypeNumber = "Number";
+    const string DataTypeString = "String";
+
+    public Amazon.SimpleNotificationService.Model.MessageAttributeValue Serialize(string key, object value) => value switch
+    {
+        // See more https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-message-metadata.html#sqs-message-attributes                    
+        var x when x is long || x is int || x is short || x is byte => new Amazon.SimpleNotificationService.Model.MessageAttributeValue
+        {
+            DataType = DataTypeNumber,
+            StringValue = value.ToString()
+        },
+        _ => new Amazon.SimpleNotificationService.Model.MessageAttributeValue
+        {
+            DataType = DataTypeString,
+            StringValue = value?.ToString()
+        }
+    };
+
+    public object Deserialize(string key, Amazon.SimpleNotificationService.Model.MessageAttributeValue value) => value.DataType switch
+    {
+        DataTypeNumber when long.TryParse(value.StringValue, out var longValue) => longValue,
+        DataTypeString when detectStringType && key != ReqRespMessageHeaders.RequestId && Guid.TryParse(value.StringValue, out var guid) => guid,
+        DataTypeString when detectStringType && bool.TryParse(value.StringValue, out var b) => b,
+        DataTypeString when detectStringType && DateTime.TryParse(value.StringValue, out var dt) => dt,
+        DataTypeString => value.StringValue,
+        _ => null
+    };
+}

src/SlimMessageBus.Host.AmazonSQS/Headers/DefaultSqsHeaderSerializer.cs

@@ -1,6 +1,6 @@
 namespace SlimMessageBus.Host.AmazonSQS;
 
-public class DefaultSqsHeaderSerializer(bool detectStringType = true) : ISqsHeaderSerializer
+public class DefaultSqsHeaderSerializer(bool detectStringType = true) : ISqsHeaderSerializer<Amazon.SQS.Model.MessageAttributeValue>
 {
     const string DataTypeNumber = "Number";
     const string DataTypeString = "String";
@@ -29,4 +29,4 @@ DataTypeNumber when long.TryParse(value.StringValue, out var longValue) => longV
         DataTypeString => value.StringValue,
         _ => null
     };
-}
+}

src/SlimMessageBus.Host.AmazonSQS/Headers/ISqsHeaderSerializer.cs

@@ -1,7 +1,8 @@
 namespace SlimMessageBus.Host.AmazonSQS;
 
-public interface ISqsHeaderSerializer
+public interface ISqsHeaderSerializer<THeaderValue> where THeaderValue : class
 {
-    MessageAttributeValue Serialize(string key, object value);
-    object Deserialize(string key, MessageAttributeValue value);
+    THeaderValue Serialize(string key, object value);
+    object Deserialize(string key, THeaderValue value);
 }
+