Add AWS SQS Support

CAP.sln

@@ -65,6 +65,8 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "DotNetCore.CAP.Test", "test
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Sample.ConsoleApp", "samples\Sample.ConsoleApp\Sample.ConsoleApp.csproj", "{2B0F467E-ABBD-4A51-BF38-D4F609DB6266}"
 EndProject
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "DotNetCore.CAP.AmazonSQS", "src\DotNetCore.CAP.AmazonSQS\DotNetCore.CAP.AmazonSQS.csproj", "{43475E00-51B7-443D-BC2D-FC21F9D8A0B4}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -147,6 +149,10 @@ Global
 		{2B0F467E-ABBD-4A51-BF38-D4F609DB6266}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{2B0F467E-ABBD-4A51-BF38-D4F609DB6266}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{2B0F467E-ABBD-4A51-BF38-D4F609DB6266}.Release|Any CPU.Build.0 = Release|Any CPU
+		{43475E00-51B7-443D-BC2D-FC21F9D8A0B4}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{43475E00-51B7-443D-BC2D-FC21F9D8A0B4}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{43475E00-51B7-443D-BC2D-FC21F9D8A0B4}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{43475E00-51B7-443D-BC2D-FC21F9D8A0B4}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
@@ -171,6 +177,7 @@ Global
 		{93176BAE-914B-4BED-9DE3-01FFB4F27FC5} = {9B2AE124-6636-4DE9-83A3-70360DABD0C4}
 		{75CC45E6-BF06-40F4-977D-10DCC05B2EFA} = {C09CDAB0-6DD4-46E9-B7F3-3EF2A4741EA0}
 		{2B0F467E-ABBD-4A51-BF38-D4F609DB6266} = {3A6B6931-A123-477A-9469-8B468B5385AF}
+		{43475E00-51B7-443D-BC2D-FC21F9D8A0B4} = {9B2AE124-6636-4DE9-83A3-70360DABD0C4}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {2E70565D-94CF-40B4-BFE1-AC18D5F736AB}

CAP.sln.DotSettings

@@ -1,4 +1,5 @@
 <wpf:ResourceDictionary xml:space="preserve" xmlns:x="http://schemas.microsoft.com/winfx/2006/xaml" xmlns:s="clr-namespace:System;assembly=mscorlib" xmlns:ss="urn:shemas-jetbrains-com:settings-storage-xaml" xmlns:wpf="http://schemas.microsoft.com/winfx/2006/xaml/presentation">
 	<s:String x:Key="/Default/CodeStyle/Naming/CSharpNaming/Abbreviations/=DB/@EntryIndexedValue">DB</s:String>
+	<s:String x:Key="/Default/CodeStyle/Naming/CSharpNaming/Abbreviations/=SNS/@EntryIndexedValue">SNS</s:String>
 	<s:Boolean x:Key="/Default/UserDictionary/Words/=Mongo/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/UserDictionary/Words/=Postgre/@EntryIndexedValue">True</s:Boolean></wpf:ResourceDictionary>

src/DotNetCore.CAP.AmazonSQS/AmazonSQSConsumerClient.cs

@@ -0,0 +1,166 @@
+// Copyright (c) .NET Core Community. All rights reserved.
+// Licensed under the MIT License. See License.txt in the project root for license information.
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading;
+using Amazon.SimpleNotificationService;
+using Amazon.SimpleNotificationService.Model;
+using Amazon.SQS;
+using Amazon.SQS.Model;
+using DotNetCore.CAP.Messages;
+using DotNetCore.CAP.Transport;
+using Microsoft.Extensions.Options;
+using Newtonsoft.Json;
+using Headers = DotNetCore.CAP.Messages.Headers;
+
+namespace DotNetCore.CAP.AmazonSQS
+{
+    internal sealed class AmazonSQSConsumerClient : IConsumerClient
+    {
+        private static readonly SemaphoreSlim ConnectionLock = new SemaphoreSlim(initialCount: 1, maxCount: 1);
+
+        private readonly string _groupId;
+        private readonly AmazonSQSOptions _amazonSQSOptions;
+
+        private IAmazonSimpleNotificationService _snsClient;
+        private IAmazonSQS _sqsClient;
+        private string _queueUrl = string.Empty;
+
+        public AmazonSQSConsumerClient(string groupId, IOptions<AmazonSQSOptions> options)
+        {
+            _groupId = groupId;
+            _amazonSQSOptions = options.Value;
+        }
+
+        public event EventHandler<TransportMessage> OnMessageReceived;
+
+        public event EventHandler<LogMessageEventArgs> OnLog;
+
+        public BrokerAddress BrokerAddress => new BrokerAddress("AmazonSQS", _queueUrl);
+
+        public void Subscribe(IEnumerable<string> topics)
+        {
+            if (topics == null)
+            {
+                throw new ArgumentNullException(nameof(topics));
+            }
+
+            Connect(initSNS: true, initSQS: false);
+
+            var topicArns = new List<string>();
+            foreach (var topic in topics)
+            {
+                var createTopicRequest = new CreateTopicRequest(topic.NormalizeForAws());
+
+                var createTopicResponse = _snsClient.CreateTopicAsync(createTopicRequest).GetAwaiter().GetResult();
+
+                topicArns.Add(createTopicResponse.TopicArn);
+            }
+
+            Connect(initSNS: false, initSQS: true);
+
+            _snsClient.SubscribeQueueToTopicsAsync(topicArns, _sqsClient, _queueUrl)
+                .GetAwaiter().GetResult();
+        }
+
+        public void Listening(TimeSpan timeout, CancellationToken cancellationToken)
+        {
+            Connect();
+
+            var request = new ReceiveMessageRequest(_queueUrl)
+            {
+                WaitTimeSeconds = 5,
+                MaxNumberOfMessages = 1
+            };
+
+            while (true)
+            {
+                var response = _sqsClient.ReceiveMessageAsync(request, cancellationToken).GetAwaiter().GetResult();
+
+                if (response.Messages.Count == 1)
+                {
+                    var messageObj = JsonConvert.DeserializeObject<SQSReceivedMessage>(response.Messages[0].Body);
+
+                    var header = messageObj.MessageAttributes.ToDictionary(x => x.Key, x => x.Value.Value);
+                    var body = messageObj.Message;
+
+                    var message = new TransportMessage(header, body != null ? Encoding.UTF8.GetBytes(body) : null);
+
+                    message.Headers.Add(Headers.Group, _groupId);
+
+                    OnMessageReceived?.Invoke(response.Messages[0].ReceiptHandle, message);
+                }
+                else
+                {
+                    cancellationToken.ThrowIfCancellationRequested();
+                    cancellationToken.WaitHandle.WaitOne(timeout);
+                }
+            }
+        }
+
+        public void Commit(object sender)
+        {
+            _sqsClient.DeleteMessageAsync(_queueUrl, (string)sender);
+        }
+
+        public void Reject(object sender)
+        {
+            _sqsClient.ChangeMessageVisibilityAsync(_queueUrl, (string)sender, 3000);
+        }
+
+        public void Dispose()
+        {
+            _sqsClient?.Dispose();
+            _snsClient?.Dispose();
+        }
+
+        public void Connect(bool initSNS = true, bool initSQS = true)
+        {
+            if (_snsClient != null && _sqsClient != null)
+            {
+                return;
+            }
+
+            if (_snsClient == null && initSNS)
+            {
+                ConnectionLock.Wait();
+
+                try
+                {
+                    _snsClient = _amazonSQSOptions.Credentials != null
+                        ? new AmazonSimpleNotificationServiceClient(_amazonSQSOptions.Credentials, _amazonSQSOptions.Region)
+                        : new AmazonSimpleNotificationServiceClient(_amazonSQSOptions.Region);
+                }
+                finally
+                {
+                    ConnectionLock.Release();
+                }
+            }
+
+            if (_sqsClient == null && initSQS)
+            {
+                ConnectionLock.Wait();
+
+                try
+                {
+
+                    _sqsClient = _amazonSQSOptions.Credentials != null
+                        ? new AmazonSQSClient(_amazonSQSOptions.Credentials, _amazonSQSOptions.Region)
+                        : new AmazonSQSClient(_amazonSQSOptions.Region);
+
+                    // If provide the name of an existing queue along with the exact names and values
+                    // of all the queue's attributes, <code>CreateQueue</code> returns the queue URL for
+                    // the existing queue.
+                    _queueUrl = _sqsClient.CreateQueueAsync(_groupId.NormalizeForAws()).GetAwaiter().GetResult().QueueUrl;
+                }
+                finally
+                {
+                    ConnectionLock.Release();
+                }
+            }
+        }
+    }
+}

src/DotNetCore.CAP.AmazonSQS/AmazonSQSConsumerClientFactory.cs

@@ -0,0 +1,31 @@
+// Copyright (c) .NET Core Community. All rights reserved.
+// Licensed under the MIT License. See License.txt in the project root for license information.
+
+using DotNetCore.CAP.Transport;
+using Microsoft.Extensions.Options;
+
+namespace DotNetCore.CAP.AmazonSQS
+{
+    internal sealed class AmazonSQSConsumerClientFactory : IConsumerClientFactory
+    {
+        private readonly IOptions<AmazonSQSOptions> _amazonSQSOptions;
+
+        public AmazonSQSConsumerClientFactory(IOptions<AmazonSQSOptions> amazonSQSOptions)
+        {
+            _amazonSQSOptions = amazonSQSOptions;
+        }
+
+        public IConsumerClient Create(string groupId)
+        {
+            try
+            {
+               var client = new AmazonSQSConsumerClient(groupId, _amazonSQSOptions);
+               return client;
+            }
+            catch (System.Exception e)
+            {
+                throw new BrokerConnectionException(e);
+            }
+        }
+    }
+}

src/DotNetCore.CAP.AmazonSQS/CAP.AmazonSQSOptions.cs

@@ -0,0 +1,17 @@
+// Copyright (c) .NET Core Community. All rights reserved.
+// Licensed under the MIT License. See License.txt in the project root for license information.
+
+using Amazon;
+using Amazon.Runtime;
+
+// ReSharper disable once CheckNamespace
+namespace DotNetCore.CAP
+{
+    // ReSharper disable once InconsistentNaming
+    public class AmazonSQSOptions
+    {
+        public RegionEndpoint Region { get; set; }
+
+        public AWSCredentials Credentials { get; set; }
+    }
+}

src/DotNetCore.CAP.AmazonSQS/CAP.AmazonSQSOptionsExtension.cs

@@ -0,0 +1,30 @@
+// Copyright (c) .NET Core Community. All rights reserved.
+// Licensed under the MIT License. See License.txt in the project root for license information.
+
+using System;
+using DotNetCore.CAP.AmazonSQS;
+using DotNetCore.CAP.Transport;
+using Microsoft.Extensions.DependencyInjection;
+
+// ReSharper disable once CheckNamespace
+namespace DotNetCore.CAP
+{
+    internal sealed class AmazonSQSOptionsExtension : ICapOptionsExtension
+    {
+        private readonly Action<AmazonSQSOptions> _configure;
+
+        public AmazonSQSOptionsExtension(Action<AmazonSQSOptions> configure)
+        {
+            _configure = configure;
+        }
+
+        public void AddServices(IServiceCollection services)
+        {
+            services.AddSingleton<CapMessageQueueMakerService>();
+
+            services.Configure(_configure);
+            services.AddSingleton<ITransport, AmazonSQSTransport>();
+            services.AddSingleton<IConsumerClientFactory, AmazonSQSConsumerClientFactory>();
+        }
+    }
+}

src/DotNetCore.CAP.AmazonSQS/CAP.Options.Extensions.cs

@@ -0,0 +1,30 @@
+// Copyright (c) .NET Core Community. All rights reserved.
+// Licensed under the MIT License. See License.txt in the project root for license information.
+
+using System;
+using Amazon;
+using DotNetCore.CAP;
+
+// ReSharper disable once CheckNamespace
+namespace Microsoft.Extensions.DependencyInjection
+{
+    public static class CapOptionsExtensions
+    {
+        public static CapOptions UseAmazonSQS(this CapOptions options, RegionEndpoint region)
+        {
+            return options.UseAmazonSQS(opt => { opt.Region =  region; });
+        }
+
+        public static CapOptions UseAmazonSQS(this CapOptions options, Action<AmazonSQSOptions> configure)
+        {
+            if (configure == null)
+            {
+                throw new ArgumentNullException(nameof(configure));
+            }
+
+            options.RegisterExtension(new AmazonSQSOptionsExtension(configure));
+
+            return options;
+        }
+    }
+}

src/DotNetCore.CAP.AmazonSQS/DotNetCore.CAP.AmazonSQS.csproj

@@ -0,0 +1,23 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <TargetFramework>netstandard2.0</TargetFramework>
+    <AssemblyName>DotNetCore.CAP.AmazonSQS</AssemblyName>
+    <PackageTags>$(PackageTags);AmazonSQS;SQS</PackageTags>
+  </PropertyGroup>
+
+  <PropertyGroup>
+    <DocumentationFile>bin\$(Configuration)\netstandard2.0\DotNetCore.CAP.AmazonSQS.xml</DocumentationFile>
+    <NoWarn>1701;1702;1705;CS1591</NoWarn>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="AWSSDK.SimpleNotificationService" Version="3.3.101.182" />
+    <PackageReference Include="AWSSDK.SQS" Version="3.3.102.125" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\DotNetCore.CAP\DotNetCore.CAP.csproj" />
+  </ItemGroup>
+
+</Project>