Improve support to AWS SQS/SNS

docker-compose-localstack.yaml

@@ -0,0 +1,15 @@
+version: '3'
+
+services:
+  localstack:
+    image: localstack/localstack
+    environment:
+      # LocalStack configuration: https://docs.localstack.cloud/references/configuration/
+      - "SERVICES=s3,sqs,sns,sts,dynamodb"
+      - "DEFAULT_REGION=eu-west-1"
+      - "DEBUG=1"
+    ports:
+      - "4566:4566" # LocalStack Gateway
+      - "4510-4559:4510-4559" # External services port range
+    volumes:
+      - "/var/run/docker.sock:/var/run/docker.sock"

samples/TaskQueue/AWSTaskQueue/Greetings/Ports/CommandHandlers/FarewellEventHandler.cs

@@ -0,0 +1,24 @@
+using System;
+using Greetings.Ports.Commands;
+using Paramore.Brighter;
+
+namespace Greetings.Ports.CommandHandlers
+{
+    public class FarewellEventHandler : RequestHandler<FarewellEvent>
+    {
+        public override FarewellEvent Handle(FarewellEvent @event)
+        {
+            Console.BackgroundColor = ConsoleColor.Blue;
+            Console.ForegroundColor = ConsoleColor.White;
+
+            Console.WriteLine("Received Farewell. Message Follows");
+            Console.WriteLine("----------------------------------");
+            Console.WriteLine(@event.Farewell);
+            Console.WriteLine("----------------------------------");
+            Console.WriteLine("Message Ends");
+
+            Console.ResetColor();
+            return base.Handle(@event);
+        }
+    }
+}

samples/TaskQueue/AWSTaskQueue/Greetings/Ports/Commands/FarewellEvent.cs

@@ -0,0 +1,19 @@
+using System;
+using Paramore.Brighter;
+
+namespace Greetings.Ports.Commands
+{
+    public class FarewellEvent : Event
+    {
+        public FarewellEvent() : base(Guid.NewGuid())
+        {
+        }
+
+        public FarewellEvent(string farewell) : base(Guid.NewGuid())
+        {
+            Farewell = farewell;
+        }
+
+        public string Farewell { get; set; }
+    }
+}

samples/TaskQueue/AWSTaskQueue/Greetings/Ports/Mappers/FarewellEventMessageMapper.cs

@@ -0,0 +1,51 @@
+#region Licence
+/* The MIT License (MIT)
+Copyright © 2014 Ian Cooper <[email protected]>
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the “Software”), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in
+all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+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.Text.Json;
+using Greetings.Ports.Commands;
+using Paramore.Brighter;
+using Paramore.Brighter.MessagingGateway.AWSSQS;
+
+namespace Greetings.Ports.Mappers
+{
+    public class FarewellEventMessageMapper : IAmAMessageMapper<FarewellEvent>
+    {
+        public IRequestContext Context { get; set; }
+
+        public Message MapToMessage(FarewellEvent request, Publication publication)
+        {
+            var header = new MessageHeader(messageId: request.Id, topic: publication.Topic, messageType: MessageType.MT_EVENT);
+            var body = new MessageBody(JsonSerializer.Serialize(request, JsonSerialisationOptions.Options));
+            var message = new Message(header, body);
+            return message;
+        }
+
+        public FarewellEvent MapToRequest(Message message)
+        {
+            var greetingCommand = JsonSerializer.Deserialize<FarewellEvent>(message.Body.Value, JsonSerialisationOptions.Options);
+
+            return greetingCommand;
+        }
+    }
+}

samples/TaskQueue/AWSTaskQueue/GreetingsPumper/Program.cs

@@ -27,32 +27,47 @@ private static async Task Main(string[] args)
             var host = new HostBuilder()
                 .ConfigureServices((hostContext, services) =>
 
-                {
-                    if (new CredentialProfileStoreChain().TryGetAWSCredentials("default", out var credentials))
                     {
-                        var awsConnection = new AWSMessagingGatewayConnection(credentials, RegionEndpoint.EUWest1);
+                        if (new CredentialProfileStoreChain().TryGetAWSCredentials("default", out var credentials))
+                        {
+                            var awsConnection = new AWSMessagingGatewayConnection(credentials, RegionEndpoint.USEast1,
+                                cfg =>
+                                {
+                                    var serviceURL = Environment.GetEnvironmentVariable("LOCALSTACK_SERVICE_URL");
+                                    if (!string.IsNullOrWhiteSpace(serviceURL))
+                                    {
+                                        cfg.ServiceURL = serviceURL;
+                                    }
+                                });
 
-                        var producerRegistry = new SnsProducerRegistryFactory(
-                            awsConnection,
-                            new SnsPublication[]
-                            {
-                                new SnsPublication
+                            var producerRegistry = new SnsProducerRegistryFactory(
+                                awsConnection,
+                                new SnsPublication[]
                                 {
-                                    Topic = new RoutingKey(typeof(GreetingEvent).FullName.ToValidSNSTopicName())
+                                    new SnsPublication
+                                    {
+                                        Topic = new RoutingKey(typeof(GreetingEvent).FullName
+                                            .ToValidSNSTopicName())
+                                    },
+                                    new SnsPublication
+                                    {
+                                        Topic = new RoutingKey(
+                                            typeof(FarewellEvent).FullName.ToValidSNSTopicName(true)),
+                                        SnsAttributes = new SnsAttributes { Type = SnsSqsType.Fifo }
+                                    }
                                 }
-                            }
-                        ).Create();
-
-                        services.AddBrighter()
-                            .UseExternalBus((configure) =>
-                            {
-                                configure.ProducerRegistry = producerRegistry;
-                            })
-                            .AutoFromAssemblies(typeof(GreetingEvent).Assembly);
-                    }
+                            ).Create();
 
-                    services.AddHostedService<RunCommandProcessor>();
-                }
+                            services.AddBrighter()
+                                .UseExternalBus((configure) =>
+                                {
+                                    configure.ProducerRegistry = producerRegistry;
+                                })
+                                .AutoFromAssemblies(typeof(GreetingEvent).Assembly);
+                        }
+
+                        services.AddHostedService<RunCommandProcessor>();
+                    }
                 )
                 .UseConsoleLifetime()
                 .UseSerilog()

samples/TaskQueue/AWSTaskQueue/GreetingsReceiverConsole/Program.cs

@@ -1,4 +1,5 @@
 #region Licence
+
 /* The MIT License (MIT)
 Copyright © 2014 Ian Cooper <[email protected]>
 
@@ -58,21 +59,36 @@ public static async Task Main(string[] args)
                             new ChannelName(typeof(GreetingEvent).FullName.ToValidSNSTopicName()),
                             new RoutingKey(typeof(GreetingEvent).FullName.ToValidSNSTopicName()),
                             bufferSize: 10,
-                            timeOut: TimeSpan.FromMilliseconds(20), 
+                            timeOut: TimeSpan.FromMilliseconds(20),
+                            lockTimeout: 30),
+                        new SqsSubscription<FarewellEvent>(new SubscriptionName("paramore.example.farewell"),
+                            new ChannelName(typeof(FarewellEvent).FullName.ToValidSNSTopicName(true)),
+                            new RoutingKey(typeof(FarewellEvent).FullName.ToValidSNSTopicName(true)),
+                            bufferSize: 10,
+                            timeOut: TimeSpan.FromMilliseconds(20),
                             lockTimeout: 30)
                     };
 
                     //create the gateway
                     if (new CredentialProfileStoreChain().TryGetAWSCredentials("default", out var credentials))
                     {
-                        var awsConnection = new AWSMessagingGatewayConnection(credentials, RegionEndpoint.EUWest1);
+                        var serviceURL = Environment.GetEnvironmentVariable("LOCALSTACK_SERVICE_URL");
+                        var region = string.IsNullOrWhiteSpace(serviceURL) ? RegionEndpoint.EUWest1 : RegionEndpoint.USEast1;
+                        var awsConnection = new AWSMessagingGatewayConnection(credentials, region,
+                            cfg =>
+                            {
+                                if (!string.IsNullOrWhiteSpace(serviceURL))
+                                {
+                                    cfg.ServiceURL = serviceURL;
+                                }
+                            });
 
                         services.AddServiceActivator(options =>
-                        {
-                            options.Subscriptions = subscriptions;
-                            options.DefaultChannelFactory = new ChannelFactory(awsConnection);
-                        })
-                        .AutoFromAssemblies();
+                            {
+                                options.Subscriptions = subscriptions;
+                                options.DefaultChannelFactory = new ChannelFactory(awsConnection);
+                            })
+                            .AutoFromAssemblies();
                     }
 
                     services.AddHostedService<ServiceActivatorHostedService>();
@@ -82,11 +98,6 @@ public static async Task Main(string[] args)
                 .Build();
 
             await host.RunAsync();
-
-
-
-
         }
     }
 }
-

samples/TaskQueue/AWSTaskQueue/GreetingsSender/Program.cs

@@ -1,4 +1,5 @@
 #region Licence
+
 /* The MIT License (MIT)
 Copyright © 2017 Ian Cooper <[email protected]>
 
@@ -22,6 +23,7 @@ THE SOFTWARE. */
 
 #endregion
 
+using System;
 using System.Transactions;
 using Amazon;
 using Amazon.Runtime.CredentialManagement;
@@ -48,10 +50,18 @@ static void Main(string[] args)
 
             var serviceCollection = new ServiceCollection();
             serviceCollection.AddSingleton<ILoggerFactory>(new SerilogLoggerFactory());
-            
+
             if (new CredentialProfileStoreChain().TryGetAWSCredentials("default", out var credentials))
             {
-                var awsConnection = new AWSMessagingGatewayConnection(credentials, RegionEndpoint.EUWest1);
+                var serviceURL = Environment.GetEnvironmentVariable("LOCALSTACK_SERVICE_URL");
+                var region = string.IsNullOrWhiteSpace(serviceURL) ? RegionEndpoint.EUWest1 : RegionEndpoint.USEast1;
+                var awsConnection = new AWSMessagingGatewayConnection(credentials, region, cfg =>
+                {
+                    if (!string.IsNullOrWhiteSpace(serviceURL))
+                    {
+                        cfg.ServiceURL = serviceURL;
+                    }
+                });
 
                 var producerRegistry = new SnsProducerRegistryFactory(
                     awsConnection,
@@ -64,7 +74,7 @@ static void Main(string[] args)
                         }
                     }
                 ).Create();
-                
+
                 serviceCollection.AddBrighter()
                     .UseExternalBus((configure) =>
                     {
@@ -76,7 +86,8 @@ static void Main(string[] args)
 
                 var commandProcessor = serviceProvider.GetService<IAmACommandProcessor>();
 
-                commandProcessor.Post(new GreetingEvent("Ian"));
+                commandProcessor.Post(new GreetingEvent("Ian says: Hi there!"));
+                commandProcessor.Post(new FarewellEvent("Ian says: See you later!"));
             }
         }
     }

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

@@ -1,4 +1,5 @@
 #region Licence
+
 /* The MIT License (MIT)
 Copyright © 2022 Ian Cooper <[email protected]>
 
@@ -31,68 +32,40 @@ THE SOFTWARE. */
 
 namespace Paramore.Brighter.MessagingGateway.AWSSQS;
 
-internal class AWSClientFactory
+internal class AWSClientFactory(
+    AWSCredentials credentials,
+    RegionEndpoint region,
+    Action<ClientConfig>? clientConfigAction)
 {
-    private readonly AWSCredentials _credentials;
-    private readonly RegionEndpoint _region;
-    private readonly Action<ClientConfig>? _clientConfigAction;
-
     public AWSClientFactory(AWSMessagingGatewayConnection connection)
+        : this(connection.Credentials, connection.Region, connection.ClientConfigAction)
     {
-        _credentials = connection.Credentials;
-        _region = connection.Region;
-        _clientConfigAction = connection.ClientConfigAction;
-    }
-
-    public AWSClientFactory(AWSCredentials credentials, RegionEndpoint region, Action<ClientConfig>? clientConfigAction)
-    {
-        _credentials = credentials;
-        _region = region;
-        _clientConfigAction = clientConfigAction;
     }
 
     public AmazonSimpleNotificationServiceClient CreateSnsClient()
     {
-        var config = new AmazonSimpleNotificationServiceConfig
-        {
-            RegionEndpoint = _region
-        };
+        var config = new AmazonSimpleNotificationServiceConfig { RegionEndpoint = region };
 
-        if (_clientConfigAction != null)
-        {
-            _clientConfigAction(config);
-        }
+        clientConfigAction?.Invoke(config);
 
-        return new AmazonSimpleNotificationServiceClient(_credentials, config);
+        return new AmazonSimpleNotificationServiceClient(credentials, config);
     }
 
     public AmazonSQSClient CreateSqsClient()
     {
-        var config = new AmazonSQSConfig
-        {
-            RegionEndpoint = _region
-        };
+        var config = new AmazonSQSConfig { RegionEndpoint = region };
 
-        if (_clientConfigAction != null)
-        {
-            _clientConfigAction(config);
-        }
+        clientConfigAction?.Invoke(config);
 
-        return new AmazonSQSClient(_credentials, config);
+        return new AmazonSQSClient(credentials, config);
     }
 
     public AmazonSecurityTokenServiceClient CreateStsClient()
     {
-        var config = new AmazonSecurityTokenServiceConfig
-        {
-            RegionEndpoint = _region
-        };
+        var config = new AmazonSecurityTokenServiceConfig { RegionEndpoint = region };
 
-        if (_clientConfigAction != null)
-        {
-            _clientConfigAction(config);
-        }
+        clientConfigAction?.Invoke(config);
 
-        return new AmazonSecurityTokenServiceClient(_credentials, config);
+        return new AmazonSecurityTokenServiceClient(credentials, config);
     }
 }