Compression and JustSaying 8.0 Changes

Directory.Build.props

@@ -60,6 +60,10 @@
     <CoverletOutputFormat>cobertura,json</CoverletOutputFormat>
     <Exclude>[*.Benchmarks]*,[*Sample*]*,[*Test*]*,[xunit.*]*</Exclude>
   </PropertyGroup>
+  <!-- Keys used by InternalsVisibleTo attributes. -->
+  <PropertyGroup>
+    <DynamicProxyGenAssembly2PublicKey>0024000004800000940000000602000000240000525341310004000001000100c547cac37abd99c8db225ef2f6c8a3602f3b3606cc9891605d02baa56104f4cfc0734aa39b93bf7852f7d9266654753cc297e7d2edfe0bac1cdcf9f717241550e0a7b191195b7667bb4f64bcb8e2121380fd1d9d46ad2d92d2d15605093924cceaf74c4861eff62abf69b9291ed0a340e113be11e6a7d3113e92484cf7045cc7</DynamicProxyGenAssembly2PublicKey>
+  </PropertyGroup>
   <ItemGroup>
     <None Include="$(MSBuildThisFileDirectory)$(PackageIcon)" Pack="True" PackagePath="" />
     <None Include="$(MSBuildThisFileDirectory)$(PackageReadmeFile)" Pack="True" PackagePath="" />

Directory.Packages.props

@@ -1,24 +1,25 @@
 <Project>
   <ItemGroup Label="Libraries">
-    <PackageVersion Include="AWSSDK.Extensions.NETCore.Setup" Version="3.7.300" />
-    <PackageVersion Include="AWSSDK.SimpleNotificationService" Version="3.7.0" />
-    <PackageVersion Include="AWSSDK.SQS" Version="3.7.0" />
-    <PackageVersion Include="Microsoft.Bcl.AsyncInterfaces" Version="1.1.1" />
+    <PackageVersion Include="AWSSDK.Extensions.NETCore.Setup" Version="3.7.301" />
+    <PackageVersion Include="AWSSDK.SimpleNotificationService" Version="3.7.400" />
+    <PackageVersion Include="AWSSDK.SQS" Version="3.7.400" />
+    <PackageVersion Include="Microsoft.Bcl.AsyncInterfaces" Version="6.0.0" />
     <PackageVersion Include="Microsoft.Extensions.Configuration" Version="8.0.0" />
     <PackageVersion Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="1.1.0" Condition=" '$(TargetFramework)' == 'net461' " />
     <PackageVersion Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="2.0.0" Condition=" '$(TargetFramework)' == 'netstandard2.0' " />
     <PackageVersion Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="8.0.0" Condition=" '$(TargetFramework)' == 'net8.0' " />
     <PackageVersion Include="Microsoft.Extensions.Logging.Abstractions" Version="2.0.0" />
     <PackageVersion Include="Newtonsoft.Json" Version="13.0.1" />
     <PackageVersion Include="StructureMap" Version="4.6.0" />
-    <PackageVersion Include="System.Text.Json" Version="4.6.0" />
+    <PackageVersion Include="System.Text.Json" Version="6.0.0" />
     <PackageVersion Include="System.Threading.Channels" Version="4.5.0" />
   </ItemGroup>
   <ItemGroup Label="Tests and Samples">
     <PackageVersion Include="AutoFixture" Version="4.18.1" />
     <PackageVersion Include="CommandLineParser" Version="2.9.1" />
     <PackageVersion Include="coverlet.msbuild" Version="6.0.2" />
     <PackageVersion Include="GitHubActionsTestLogger" Version="2.4.1" />
+    <PackageVersion Include="LocalSqsSnsMessaging" Version="0.5.2" />
     <PackageVersion Include="Magnum" Version="2.1.3" />
     <PackageVersion Include="MartinCostello.Logging.XUnit" Version="0.4.0" />
     <PackageVersion Include="MELT" Version="0.9.0" />

samples/src/JustSaying.Sample.Restaurant.KitchenConsole/Program.cs

@@ -1,3 +1,5 @@
+using JustSaying.AwsTools.MessageHandling;
+using JustSaying.Messaging.Compression;
 using JustSaying.Sample.Restaurant.KitchenConsole;
 using JustSaying.Sample.Restaurant.KitchenConsole.Handlers;
 using JustSaying.Sample.Restaurant.Models;
@@ -97,13 +99,19 @@ static async Task Run()
                 config.Publications(x =>
                 {
                     // Creates the following if they do not already exist
-                    //  - a SNS topic of name `orderreadyevent` with two tags:
+                    //  - an SNS topic of name `orderreadyevent` with two tags:
                     //      - "IsOrderEvent" with no value
                     //      - "Publisher" with the value "KitchenConsole"
                     x.WithTopic<OrderReadyEvent>(cfg =>
                     {
                         cfg.WithTag("IsOrderEvent")
-                            .WithTag("Publisher", appName);
+                            .WithTag("Publisher", appName)
+                            .WithWriteConfiguration(w =>
+                                w.CompressionOptions = new PublishCompressionOptions
+                                {
+                                    CompressionEncoding = ContentEncodings.GzipBase64, // TODO validate
+                                    MessageLengthThreshold = 0
+                                });
                     });
                     x.WithTopic<OrderDeliveredEvent>();
                 });

samples/src/JustSaying.Sample.Restaurant.KitchenConsole/appsettings.json

@@ -1,4 +1,4 @@
 {
   "AWSRegion": "eu-west-1",
-  "AWSServiceUrl": "http://localhost:4100"
+  "AWSServiceUrl": "http://localhost:4566"
 }

samples/src/JustSaying.Sample.Restaurant.OrderingApi/appsettings.json

@@ -1,5 +1,5 @@
 {
   "AWSRegion": "eu-west-1",
-  "AWSServiceUrl": "http://localhost:4100",
+  "AWSServiceUrl": "http://localhost:4566",
   "AllowedHosts": "*"
 }

src/JustSaying.Extensions.DependencyInjection.Microsoft/IServiceCollectionExtensions.cs

@@ -4,6 +4,7 @@
 using JustSaying.AwsTools.QueueCreation;
 using JustSaying.Fluent;
 using JustSaying.Messaging.Channels.Receive;
+using JustSaying.Messaging.Compression;
 using JustSaying.Messaging.MessageHandling;
 using JustSaying.Messaging.MessageSerialization;
 using JustSaying.Messaging.Middleware.Logging;
@@ -135,16 +136,10 @@ public static IServiceCollection AddJustSaying(this IServiceCollection services,
         services.TryAddSingleton<IMessageContextAccessor>(serviceProvider => serviceProvider.GetRequiredService<MessageContextAccessor>());
         services.TryAddSingleton<IMessageContextReader>(serviceProvider => serviceProvider.GetRequiredService<MessageContextAccessor>());
 
-        services.TryAddSingleton<IMessageSerializationFactory, NewtonsoftSerializationFactory>();
+        services.TryAddSingleton<IMessageBodySerializationFactory, NewtonsoftSerializationFactory>();
         services.TryAddSingleton<IMessageSubjectProvider, GenericMessageSubjectProvider>();
         services.TryAddSingleton<IVerifyAmazonQueues, AmazonQueueCreator>();
-        services.TryAddSingleton<IMessageSerializationRegister>(
-            (p) =>
-            {
-                var config = p.GetRequiredService<IMessagingConfig>();
-                var serializerFactory = p.GetRequiredService<IMessageSerializationFactory>();
-                return new MessageSerializationRegister(config.MessageSubjectProvider, serializerFactory);
-            });
+        services.TryAddSingleton((p) => new MessageCompressionRegistry([new GzipMessageBodyCompression()]));
 
         services.TryAddSingleton<IMessageReceivePauseSignal, MessageReceivePauseSignal>();
 

src/JustSaying.Extensions.DependencyInjection.StructureMap/JustSayingRegistry.cs

@@ -2,6 +2,7 @@
 using JustSaying.AwsTools.QueueCreation;
 using JustSaying.Fluent;
 using JustSaying.Messaging.Channels.Receive;
+using JustSaying.Messaging.Compression;
 using JustSaying.Messaging.MessageHandling;
 using JustSaying.Messaging.MessageSerialization;
 using JustSaying.Messaging.Middleware.Logging;
@@ -31,7 +32,7 @@ public JustSayingRegistry()
         For<IAwsClientFactoryProxy>().Use((p) => new AwsClientFactoryProxy(p.GetInstance<IAwsClientFactory>)).Singleton();
         For<IMessagingConfig>().Use<MessagingConfig>().Singleton();
         For<IMessageMonitor>().Use<NullOpMessageMonitor>().Singleton();
-        For<IMessageSerializationFactory>().Use<NewtonsoftSerializationFactory>().Singleton();
+        For<IMessageBodySerializationFactory>().Use<NewtonsoftSerializationFactory>().Singleton();
         For<IMessageSubjectProvider>().Use<GenericMessageSubjectProvider>().Singleton();
         For<IVerifyAmazonQueues>().Use<AmazonQueueCreator>().Singleton();
 
@@ -41,18 +42,7 @@ public JustSayingRegistry()
 
         For<LoggingMiddleware>().Transient();
         For<SqsPostProcessorMiddleware>().Transient();
-
-        For<IMessageSerializationRegister>()
-            .Use(
-                nameof(IMessageSerializationRegister),
-                (p) =>
-                {
-                    var config = p.GetInstance<IMessagingConfig>();
-                    var serializerFactory = p.GetInstance<IMessageSerializationFactory>();
-                    return new MessageSerializationRegister(config.MessageSubjectProvider, serializerFactory);
-                })
-            .Singleton();
-
+        For<MessageCompressionRegistry>().Use((p) => new MessageCompressionRegistry(new List<IMessageBodyCompression> { new GzipMessageBodyCompression() })).Singleton();
         For<IMessageReceivePauseSignal>().Use<MessageReceivePauseSignal>().Singleton();
 
         For<DefaultNamingConventions>().Singleton();

src/JustSaying/AwsTools/MessageAttributeKeys.cs

@@ -0,0 +1,12 @@
+namespace JustSaying.AwsTools;
+
+/// <summary>
+/// Contains constant key values for message attributes.
+/// </summary>
+internal static class MessageAttributeKeys
+{
+    /// <summary>
+    /// Represents the key for the Content-Encoding attribute.
+    /// </summary>
+    public const string ContentEncoding = "Content-Encoding";
+}

src/JustSaying/AwsTools/MessageHandling/Compression/PublishCompressionOptions.cs

@@ -0,0 +1,23 @@
+namespace JustSaying.AwsTools.MessageHandling;
+
+/// <summary>
+/// Represents options for message compression during publishing.
+/// </summary>
+public sealed class PublishCompressionOptions
+{
+    /// <summary>
+    /// Gets or sets the message length threshold in bytes.
+    /// Messages larger than this threshold will be compressed.
+    /// </summary>
+    /// <remarks>
+    /// The default value is 262,144 bytes (256 KB).
+    /// </remarks>
+    public int MessageLengthThreshold { get; set; } = 256 * 1024;
+    /// <summary>
+    /// Gets or sets the compression encoding to be used.
+    /// </summary>
+    /// <remarks>
+    /// This should correspond to a registered compression algorithm in the IMessageCompressionRegistry.
+    /// </remarks>
+    public string CompressionEncoding { get; set; }
+}

src/JustSaying/AwsTools/MessageHandling/Dispatch/IMessageDispatcher.cs

@@ -5,7 +5,7 @@ namespace JustSaying.AwsTools.MessageHandling.Dispatch;
 /// <summary>
 /// Dispatches messages to the queue.
 /// </summary>
-public interface IMessageDispatcher
+internal interface IMessageDispatcher
 {
     /// <summary>
     /// Dispatches the message in <see cref="IQueueMessageContext"/> to the queue in the context.
@@ -14,4 +14,4 @@ public interface IMessageDispatcher
     /// <param name="cancellationToken">A <see cref="CancellationToken"/> to stop processing the message dispatch.</param>
     /// <returns>A <see cref="Task"/> that completes once the message has been dispatched.</returns>
     Task DispatchMessageAsync(IQueueMessageContext messageContext, CancellationToken cancellationToken);
-}
+}

src/JustSaying/AwsTools/MessageHandling/Dispatch/MessageDispatcher.cs

@@ -8,21 +8,18 @@
 
 namespace JustSaying.AwsTools.MessageHandling.Dispatch;
 
-public class MessageDispatcher : IMessageDispatcher
+internal class MessageDispatcher : IMessageDispatcher
 {
-    private readonly IMessageSerializationRegister _serializationRegister;
     private readonly IMessageMonitor _messagingMonitor;
     private readonly MiddlewareMap _middlewareMap;
 
     private static ILogger _logger;
 
     public MessageDispatcher(
-        IMessageSerializationRegister serializationRegister,
         IMessageMonitor messagingMonitor,
         MiddlewareMap middlewareMap,
         ILoggerFactory loggerFactory)
     {
-        _serializationRegister = serializationRegister;
         _messagingMonitor = messagingMonitor;
         _middlewareMap = middlewareMap;
         _logger = loggerFactory.CreateLogger("JustSaying");
@@ -68,18 +65,18 @@ public async Task DispatchMessageAsync(
 
         await middleware.RunAsync(handleContext, null, cancellationToken)
             .ConfigureAwait(false);
-
     }
 
     private async Task<(bool success, Message typedMessage, MessageAttributes attributes)>
         DeserializeMessage(IQueueMessageContext messageContext, CancellationToken cancellationToken)
     {
         try
         {
-            _logger.LogDebug("Attempting to deserialize message with serialization register {Type}",
-                _serializationRegister.GetType().FullName);
-            var messageWithAttributes = _serializationRegister.DeserializeMessage(messageContext.Message.Body);
-            return (true, messageWithAttributes.Message, messageWithAttributes.MessageAttributes);
+            _logger.LogDebug("Attempting to deserialize message.");
+
+            var (message, attributes) = messageContext.MessageConverter.ConvertForReceive(messageContext.Message);
+
+            return (true, message, attributes);
         }
         catch (MessageFormatNotSupportedException ex)
         {

src/JustSaying/AwsTools/MessageHandling/ForeignTopicArnProvider.cs

@@ -2,17 +2,10 @@
 
 namespace JustSaying.AwsTools.MessageHandling;
 
-internal class ForeignTopicArnProvider(RegionEndpoint regionEndpoint, string accountId, string topicName) : ITopicArnProvider
+internal class ForeignTopicArnProvider(RegionEndpoint regionEndpoint, string accountId, string topicName)
 {
-
     private readonly string _arn = $"arn:aws:sns:{regionEndpoint.SystemName}:{accountId}:{topicName}";
 
-    public Task<bool> ArnExistsAsync()
-    {
-        // Assume foreign topics exist, we actually find out when we attempt to subscribe
-        return Task.FromResult(true);
-    }
-
     public Task<string> GetArnAsync()
     {
         return Task.FromResult(_arn);

src/JustSaying/AwsTools/MessageHandling/IMessageConverter.cs

@@ -0,0 +1,15 @@
+using JustSaying.AwsTools.MessageHandling;
+using JustSaying.Messaging.MessageSerialization;
+using JustSaying.Models;
+
+namespace JustSaying.Messaging;
+
+public interface IPublishMessageConverter
+{
+    PublishMessage ConvertForPublish(Message message, PublishMetadata publishMetadata, PublishDestinationType destinationType);
+}
+
+public interface IReceivedMessageConverter
+{
+    ReceivedMessage ConvertForReceive(Amazon.SQS.Model.Message message);
+}

src/JustSaying/AwsTools/MessageHandling/ISqsQueue.cs

@@ -76,4 +76,4 @@ Task ChangeMessageVisibilityAsync(
         int visibilityTimeoutInSeconds,
         CancellationToken cancellationToken);
 
-}
+}

src/JustSaying/AwsTools/MessageHandling/PublishDestinationType.cs

@@ -0,0 +1,7 @@
+namespace JustSaying.AwsTools.MessageHandling;
+
+public enum PublishDestinationType
+{
+    Topic = 1,
+    Queue = 2
+}