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,32 +1,32 @@
 <Project>
   <ItemGroup Label="Libraries">
-    <PackageVersion Include="AWSSDK.Extensions.NETCore.Setup" Version="3.7.300" />
+    <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="1.1.1" />
+    <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.10" />
     <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.3" />
     <PackageVersion Include="Magnum" Version="2.1.3" />
     <PackageVersion Include="MartinCostello.Logging.XUnit" Version="0.4.0" />
-    <PackageVersion Include="MELT" Version="0.9.0" />
-    <PackageVersion Include="MELT.Xunit" Version="0.9.0" />
     <PackageVersion Include="Microsoft.CodeAnalysis.PublicApiAnalyzers" Version="3.3.4" />
     <PackageVersion Include="Microsoft.Extensions.Configuration.EnvironmentVariables" Version="8.0.0" />
     <PackageVersion Include="Microsoft.Extensions.Configuration.Json" Version="8.0.1" />
     <PackageVersion Include="Microsoft.Extensions.DependencyInjection" Version="8.0.1" />
+    <PackageVersion Include="Microsoft.Extensions.Diagnostics.Testing" Version="8.10.0" />
     <PackageVersion Include="Microsoft.Extensions.Hosting" Version="8.0.1" />
     <PackageVersion Include="Microsoft.Extensions.Logging" Version="8.0.1" />
     <PackageVersion Include="Microsoft.Extensions.Logging.Console" Version="8.0.1" />

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

@@ -97,7 +97,7 @@ 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 =>

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

@@ -5,6 +5,7 @@
 using JustSaying.Fluent;
 using JustSaying.Messaging;
 using JustSaying.Messaging.Channels.Receive;
+using JustSaying.Messaging.Compression;
 using JustSaying.Messaging.MessageHandling;
 using JustSaying.Messaging.MessageSerialization;
 using JustSaying.Messaging.Middleware.Logging;
@@ -139,16 +140,11 @@ 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.TryAddEnumerable(ServiceDescriptor.Singleton<IMessageBodyCompression, GzipMessageBodyCompression>());
+        services.TryAddSingleton<MessageCompressionRegistry>();
 
         services.TryAddSingleton<IMessageReceivePauseSignal, MessageReceivePauseSignal>();
 

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

@@ -2,13 +2,15 @@
 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;
 using JustSaying.Messaging.Middleware.PostProcessing;
 using JustSaying.Messaging.Monitoring;
 using JustSaying.Naming;
 using StructureMap;
+using StructureMap.Pipeline;
 
 namespace JustSaying;
 
@@ -33,7 +35,7 @@ public JustSayingRegistry()
         For<IMessagingConfig>().Use(context => context.GetInstance<MessagingConfig>()).Singleton();
         For<IPublishBatchConfiguration>().Use<MessagingConfig>(context => context.GetInstance<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();
 
@@ -43,18 +45,8 @@ 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<IMessageBodyCompression>().Add<GzipMessageBodyCompression>().Singleton();
+        For<MessageCompressionRegistry>().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,26 @@
+using JustSaying.Messaging.Compression;
+
+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 260,096 bytes (254 KB), 2KB less than the SNS and SQS limit.
+    /// </remarks>
+    public int MessageLengthThreshold { get; set; } = 254 * 1024;
+
+    /// <summary>
+    /// Gets or sets the compression encoding to be used.
+    /// </summary>
+    /// <remarks>
+    /// This should correspond to a registered compression algorithm in the <see cref="MessageCompressionRegistry"/>.
+    /// </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 sealed class MessageDispatcher : IMessageDispatcher
 {
-    private readonly IMessageSerializationRegister _serializationRegister;
     private readonly IMessageMonitor _messagingMonitor;
     private readonly MiddlewareMap _middlewareMap;
 
-    private static ILogger _logger;
+    private readonly ILogger _logger;
 
     public MessageDispatcher(
-        IMessageSerializationRegister serializationRegister,
         IMessageMonitor messagingMonitor,
         MiddlewareMap middlewareMap,
         ILoggerFactory loggerFactory)
     {
-        _serializationRegister = serializationRegister;
         _messagingMonitor = messagingMonitor;
         _middlewareMap = middlewareMap;
         _logger = loggerFactory.CreateLogger("JustSaying");
@@ -42,6 +39,7 @@
 
         if (!success)
         {
+            _logger.LogTrace("DeserializeMessage failed. Message will not be dispatched.");
             return;
         }
 
@@ -68,18 +66,18 @@
 
         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) = await messageContext.MessageConverter.ConvertToInboundMessageAsync(messageContext.Message, cancellationToken);
+
+            return (true, message, attributes);
         }
         catch (MessageFormatNotSupportedException ex)
         {
@@ -93,6 +91,11 @@
 
             return (false, null, null);
         }
+        catch (OperationCanceledException)
+        {
+            // Ignore cancellation
+            return (false, null, null);
+        }
         catch (Exception ex)
         {
             _logger.LogError(

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/IInboundMessageConverter.cs

@@ -0,0 +1,18 @@
+using JustSaying.Messaging.MessageSerialization;
+
+namespace JustSaying.Messaging;
+
+public interface IInboundMessageConverter
+{
+    /// <summary>
+    /// Converts an Amazon SQS message to a <see cref="InboundMessage" /> object.
+    /// </summary>
+    /// <param name="message">The Amazon SQS message to convert.</param>
+    /// <param name="cancellationToken">A token to monitor for cancellation requests.</param>
+    /// <returns>A <see cref="InboundMessage" /> object containing the deserialized message body and attributes.</returns>
+    /// <remarks>
+    /// This method handles the conversion of both raw SQS messages and SNS-wrapped messages.
+    /// It also applies any necessary decompression to the message body.
+    /// </remarks>
+    ValueTask<InboundMessage> ConvertToInboundMessageAsync(Amazon.SQS.Model.Message message, CancellationToken cancellationToken = default);
+}

src/JustSaying/AwsTools/MessageHandling/IOutboundMessageConverter.cs

@@ -0,0 +1,27 @@
+using JustSaying.Messaging.MessageSerialization;
+using JustSaying.Models;
+
+namespace JustSaying.Messaging;
+
+public interface IOutboundMessageConverter
+{
+    /// <summary>
+    /// Converts a message to a format suitable for publishing, applying necessary transformations and compression.
+    /// </summary>
+    /// <param name="message">The original message to be converted.</param>
+    /// <param name="publishMetadata">Metadata associated with the publish operation, including any custom message attributes.</param>
+    /// <param name="cancellationToken">A token to monitor for cancellation requests.</param>
+    /// <returns>A <see cref="OutboundMessage"/> object containing the converted message body, attributes, and any additional publishing information.</returns>
+    /// <remarks>
+    /// This method handles the following operations:
+    /// <ul>
+    /// <li>Serializes the message body</li>
+    /// <li>Adds custom message attributes</li>
+    /// <li>Applies compression to the message body if it meets specified criteria</li>
+    /// <li>Adds compression-related attributes if compression is applied</li>
+    /// <li>Prepares the message for SNS (if applicable) by setting the subject</li>
+    /// </ul>
+    /// The exact behavior may vary based on the destination type and compression options.
+    /// </remarks>
+    ValueTask<OutboundMessage> ConvertToOutboundMessageAsync(Message message, PublishMetadata publishMetadata, CancellationToken cancellationToken = default);
+}

src/JustSaying/AwsTools/MessageHandling/ISqsQueue.cs

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